diff --git a/.env.example b/.env.example index 6d127382479..d05d4c5f109 100644 --- a/.env.example +++ b/.env.example @@ -69,6 +69,11 @@ RELAY_URL=ws://localhost:3000 # Stable relay signing key (required). `just bootstrap` generates a random key in # the gitignored .env file. Preserve that value across restarts and backups. # BUZZ_RELAY_PRIVATE_KEY=<32-byte hex private key> +# Host-wide NIP-11 workflow lifecycle advertisement: default off; only true/1 opts in. +# Enable only AFTER all incompatible endpoints AND existing connections are drained. +# Gates discovery only, not lifecycle enforcement. See the two-phase operator sequence: +# deploy/charts/buzz/README.md#workflow-lifecycle-activation +BUZZ_ADVERTISE_WORKFLOW_LIFECYCLE=false # Optional: path to the web UI dist directory. When set, the relay serves # the web frontend at / for browser requests. Leave unset for local dev # (use `just web` for Vite HMR instead). diff --git a/crates/buzz-agent/tests/fake_llm.rs b/crates/buzz-agent/tests/fake_llm.rs index 9822243d5fe..543392f26a6 100644 --- a/crates/buzz-agent/tests/fake_llm.rs +++ b/crates/buzz-agent/tests/fake_llm.rs @@ -1320,7 +1320,7 @@ async fn mid_turn_usage_includes_earlier_turns() { /// Setup: round 1 is a tool call WITH usage (tokens are captured). After the /// tool_call_update notification (proving round 1 is fully processed), we gate /// the round-2 LLM response behind a `oneshot` barrier that only releases after -/// cancel is sent. This guarantees the turn exits with `stopReason: "cancelled"` +/// cancel is acknowledged. This guarantees the turn exits with `stopReason: "cancelled"` /// deterministically, even on a slow CI worker. #[tokio::test(flavor = "multi_thread", worker_threads = 2)] async fn cancelled_turn_with_usage_emits_notification_before_response() { @@ -1331,11 +1331,9 @@ async fn cancelled_turn_with_usage_emits_notification_before_response() { let gate_rx = Arc::new(tokio::sync::Mutex::new(Some(gate_rx))); // Round 1: tool call with usage — sets turn_input/output_tokens. - // Round 2: gated — blocked until cancel fires, then released so the - // in-flight TCP request can resolve. The queue is empty for round 2, so the - // agent receives the fallback "no canned response" body which it treats as - // an LLM error; the cancel check at the round boundary fires first because - // the gate is only released after cancel is enqueued. + // Round 2 stays blocked until the agent acknowledges cancellation. Writing + // cancel to stdin is not a barrier: the HTTP error could win before the + // agent reads stdin. handle_request sends the ack after cancel_session. let responses = vec![openai_tool_call_with_usage( "call_cancel_test", "fake__noop", @@ -1371,7 +1369,7 @@ async fn cancelled_turn_with_usage_emits_notification_before_response() { } } // For request 2+ (round 2), wait for the gate to open before - // responding. This ensures cancel is sent before round 2 resolves, + // responding. This ensures cancel is acknowledged before round 2 resolves, // making stopReason: cancelled deterministic. if req_num >= 2 { let rx = gate.lock().await.take(); @@ -1415,10 +1413,10 @@ async fn cancelled_turn_with_usage_emits_notification_before_response() { }) .await; - // Now send cancel and release the round-2 gate. Cancel is enqueued before - // round 2 can respond, so the turn exits with stopReason: cancelled. + // Collect every message while waiting for the ack: usage and the prompt + // response can precede it, so recv_until would discard evidence of ordering. let c_id = h.send("session/cancel", json!({"sessionId": sid})).await; - let _ = gate_tx.send(()); // unblock round 2 + let mut gate_tx = Some(gate_tx); let mut saw_usage_before_prompt_response = false; let mut saw_usage = false; @@ -1427,7 +1425,12 @@ async fn cancelled_turn_with_usage_emits_notification_before_response() { for _ in 0..40 { let v = h.recv().await; if v["id"] == json!(c_id) { + assert_eq!(v.get("result"), Some(&Value::Null), "cancel must succeed"); + assert!(v.get("error").is_none(), "cancel must not return an error"); saw_cancel_ok = true; + if let Some(gate_tx) = gate_tx.take() { + let _ = gate_tx.send(()); // cancellation is now installed + } } else if is_usage_update(&v) { saw_usage = true; if !saw_prompt_response { @@ -1446,6 +1449,7 @@ async fn cancelled_turn_with_usage_emits_notification_before_response() { } } assert!(saw_cancel_ok, "session/cancel was not acknowledged"); + assert!(saw_prompt_response, "session/prompt did not complete"); assert!( saw_usage, "expected usage_update notification for cancelled turn with observed tokens" diff --git a/crates/buzz-db/src/runtime/migration.rs b/crates/buzz-db/src/runtime/migration.rs index 59015125042..0ac2da029b9 100644 --- a/crates/buzz-db/src/runtime/migration.rs +++ b/crates/buzz-db/src/runtime/migration.rs @@ -702,7 +702,7 @@ mod postgres_tests { let mut migrations: Vec<_> = MIGRATOR.iter().collect(); migrations.sort_by_key(|migration| migration.version); - assert_eq!(migrations.len(), 44); + assert_eq!(migrations.len(), 45); assert_eq!(migrations[0].version, 1); assert_eq!(&*migrations[0].description, "initial schema"); assert!(migrations[0] @@ -1368,6 +1368,35 @@ mod postgres_tests { assert!(include_str!("../../../../schema/schema.sql").contains("error_code TEXT")); } + #[test] + fn workflow_deletion_cutoff_matches_desired_schema_without_legacy_backfill() { + let migration = MIGRATOR + .iter() + .find(|m| m.version == 45) + .expect("migration 45"); + let schema = include_str!("../../../../schema/schema.sql"); + let table = |sql: &str| { + split_sql_statements(sql) + .into_iter() + .find(|statement| { + normalize_sql(statement).starts_with("create table workflow_deletions") + }) + .map(|statement| normalize_sql(&statement)) + .expect("deletion table") + }; + assert_eq!(table(migration.sql.as_ref()), table(schema)); + let statements = split_sql_statements(migration.sql.as_ref()); + assert_eq!( + statements.len(), + 2, + "forward proof must never be inferred from legacy data" + ); + assert_eq!( + normalize_sql(&statements[1]), + "select attach_community_write_fence('workflow_deletions')" + ); + } + #[test] fn push_match_trigger_is_narrowed_to_message_kinds_additively() { let mut migrations: Vec<_> = MIGRATOR.iter().collect(); @@ -1822,6 +1851,11 @@ mod postgres_tests { let mut expected_fences = migration.fence_attachments.clone(); expected_fences.remove("product_feedback"); expected_fences.remove("rate_limit_violations"); + let workflow_lifecycle = MIGRATOR + .iter() + .find(|migration| migration.version == 45) + .expect("embedded workflow lifecycle migration"); + expected_fences.extend(surface(workflow_lifecycle.sql.as_ref()).fence_attachments); assert_eq!( expected_fences, schema.fence_attachments, "write-fence attachment targets differ after recovery policy" @@ -2752,10 +2786,16 @@ mod postgres_tests { "all NIP-FI tables must be absent after migration 0044: {present:?}" ); - // The deletion catalog must validate with ledger relations gone. + // Ledger absence above is checked at 0044. The current binary's deletion + // catalog describes the latest schema (including later scoped tables), + // so finish the upgrade before asking it to validate serving readiness. + MIGRATOR + .run(&pool) + .await + .expect("finish migrations after ledger removal"); crate::deletion::DeletionStore::new(pool.clone()) .validate_catalog() .await - .expect("deletion catalog validates after migration 0044"); + .expect("current deletion catalog validates after upgrading from migration 0044"); } } diff --git a/crates/buzz-db/src/store/deletion.rs b/crates/buzz-db/src/store/deletion.rs index 0e184e00d88..dd316b313dc 100644 --- a/crates/buzz-db/src/store/deletion.rs +++ b/crates/buzz-db/src/store/deletion.rs @@ -81,12 +81,14 @@ pub const EXPECTED_SCOPED_TABLES: &[&str] = &[ "thread_metadata", "users", "workflow_approvals", + "workflow_deletions", "workflow_runs", "workflows", ]; /// Foreign-key-safe child-before-parent order for the PostgreSQL purge. pub const PURGE_SCOPED_TABLES: &[&str] = &[ + "workflow_deletions", "workflow_approvals", "scheduled_workflow_fires", "workflow_runs", @@ -4997,3 +4999,7 @@ mod postgres_tests { .expect("drop probe database"); } } + +#[cfg(test)] +#[path = "deletion/workflow_postgres_tests.rs"] +mod workflow_postgres_tests; diff --git a/crates/buzz-db/src/store/deletion/workflow_postgres_tests.rs b/crates/buzz-db/src/store/deletion/workflow_postgres_tests.rs new file mode 100644 index 00000000000..dbc6a18e673 --- /dev/null +++ b/crates/buzz-db/src/store/deletion/workflow_postgres_tests.rs @@ -0,0 +1,158 @@ +//! The forward workflow proof is tenant data, including during whole-community purge. +use super::*; +use buzz_core::channel::{ChannelType, ChannelVisibility}; + +#[tokio::test] +#[ignore = "requires Postgres"] +async fn desired_workflow_deletion_proofs_are_fenced_and_purged() { + check_workflow_deletion_proofs("desired").await; +} + +#[tokio::test] +#[ignore = "requires Postgres"] +async fn migration_schema_workflow_deletion_proofs_are_fenced_and_purged() { + check_workflow_deletion_proofs("migration").await; +} + +async fn check_workflow_deletion_proofs(mode: &str) { + assert_eq!(std::env::var("BUZZ_TEST_SCHEMA_MODE").as_deref(), Ok(mode)); + let pool = PgPool::connect(&crate::test_support::database_url()) + .await + .expect("isolated database"); + let db = Db::from_pool(pool.clone()); + if mode == "migration" { + db.migrate().await.expect("all migrations"); + } + let store = db.deletion_store(); + let mut tenants = Vec::new(); + for label in ["target", "control"] { + let host = format!("workflow-purge-{label}-{}.example", Uuid::new_v4()); + let community = db + .ensure_configured_community(&host) + .await + .expect("community") + .id; + let channel = db + .create_channel( + community, + "workflows", + ChannelType::Stream, + ChannelVisibility::Private, + None, + &[7; 32], + None, + ) + .await + .expect("channel") + .id; + sqlx::query( + "INSERT INTO workflow_deletions \ + (community_id, owner_pubkey, workflow_id, channel_id, deleted_through, event_id) \ + VALUES ($1, $2, $3, $4, now(), $5)", + ) + .bind(community.as_uuid()) + .bind(vec![7_u8; 32]) + .bind(Uuid::new_v4()) + .bind(channel) + .bind(vec![8_u8; 32]) + .execute(&pool) + .await + .expect("proof"); + tenants.push((community, host)); + } + let (target, host) = &tenants[0]; + let schema = store + .inventory_schema(*target) + .await + .expect("catalog includes proof"); + assert_eq!(schema.row_counts["workflow_deletions"], 1); + assert!(schema + .fenced_tables + .contains(&"workflow_deletions".to_string())); + let storage = StorageManifest { + version: 5, + prefixes: ["_meta", "_uploads", "repos"] + .into_iter() + .map(|prefix| { + let (keys_digest, object_count) = KeyStreamDigest::new().finish(); + PrefixManifest { + prefix: format!("{prefix}/{target}/"), + keys_digest, + object_count, + total_bytes: 0, + } + }) + .collect(), + }; + let request = store + .submit(host, "test-operator", None) + .await + .expect("request"); + store + .freeze_inventory( + request.id, + &FrozenInventory { + schema, + storage: storage.clone(), + }, + ) + .await + .expect("freeze"); + store + .approve(request.id, "test-approver", None) + .await + .expect("approve"); + let claim = store + .claim_specific(request.id, "test-executor", DEFAULT_LEASE_DURATION) + .await + .expect("claim") + .expect("won claim"); + store.begin_quiescing(&claim.lease).await.expect("quiesce"); + let generation = store.fence(&claim.lease).await.expect("fence"); + let token = LeaseToken { + fence_generation: Some(generation), + ..claim.lease + }; + let error = sqlx::query( + "UPDATE workflow_deletions SET deleted_through = now() WHERE community_id = $1", + ) + .bind(target.as_uuid()) + .execute(&pool) + .await + .expect_err("proof updates must be fenced"); + assert!( + error.to_string().contains("community write fenced"), + "{error}" + ); + store + .freeze_destructive_storage_manifest(&token, &storage) + .await + .expect("storage"); + store.mark_drained(&token).await.expect("drain"); + store + .mark_bindings_removed(&token, serde_json::json!({"keys": 0})) + .await + .expect("bindings"); + let counts = store + .purge_postgres(&token) + .await + .expect("purge child before channel"); + assert_eq!(counts["workflow_deletions"], 1); + store + .mark_cache_purged(&token, serde_json::json!({"keys": 0})) + .await + .expect("cache"); + store + .verify_postgres_logically_deleted(&token) + .await + .expect("logical absence"); + let remaining: Vec = sqlx::query_scalar("SELECT community_id FROM workflow_deletions") + .fetch_all(&pool) + .await + .expect("remaining proofs"); + assert_eq!( + remaining, + vec![*tenants[1].0.as_uuid()], + "control tenant stays intact" + ); +} diff --git a/crates/buzz-db/src/store/relay_admin_actions.rs b/crates/buzz-db/src/store/relay_admin_actions.rs index 7662077911d..45595e12fb2 100644 --- a/crates/buzz-db/src/store/relay_admin_actions.rs +++ b/crates/buzz-db/src/store/relay_admin_actions.rs @@ -693,8 +693,9 @@ pub async fn execute_kick_with_marker( /// Atomically execute a soft-delete mutation and commit the step marker in one /// transaction, fenced by `action_id` AND the caller's `lease_token`. /// -/// The delete is idempotent: if the event is already deleted the marker is still -/// committed (soft-delete is already-done = success). +/// Workflow definitions are rejected, including already-deleted ones; they require +/// canonical author-signed lifecycle deletion. For other events, an already-deleted +/// target is idempotent success and the marker is still committed. pub async fn execute_delete_with_marker( pool: &PgPool, action_id: Uuid, @@ -727,6 +728,25 @@ pub async fn execute_delete_with_marker( return Ok(false); } + // Workflow definitions are one projection of executable state. This shared + // boundary also serves recovery-worker retries, so HTTP-only validation is + // insufficient. Reject tombstoned definitions too: they do not make a new + // moderation action an author-signed lifecycle deletion. Event kind is immutable. + let workflow: bool = sqlx::query_scalar( + "SELECT EXISTS (SELECT 1 FROM events WHERE community_id = $1 AND id = $2 AND kind = $3)", + ) + .bind(community_id.as_uuid()) + .bind(target_event_id) + .bind(buzz_core::kind::KIND_WORKFLOW_DEF as i32) + .fetch_one(&mut *tx) + .await?; + if workflow { + tx.rollback().await?; + return Err(crate::DbError::InvalidData( + "workflow definitions require canonical author-signed deletion".into(), + )); + } + // Soft-delete the event and update thread metadata (idempotent: already-deleted is a no-op). sqlx::query( r#" diff --git a/crates/buzz-db/src/store/workflow.rs b/crates/buzz-db/src/store/workflow.rs index 3ceed9ea32e..f3e3b59ba01 100644 --- a/crates/buzz-db/src/store/workflow.rs +++ b/crates/buzz-db/src/store/workflow.rs @@ -7,12 +7,18 @@ //! - Approval tokens are stored as SHA-256 hashes (never plaintext). //! - All list queries have a bounded LIMIT to prevent unbounded scans. +/// Transactional workflow-coordinate lifecycle operations. +pub mod lifecycle; + +mod run_admission; +pub use run_admission::create_workflow_run; + use std::fmt; use std::str::FromStr; use chrono::{DateTime, Utc}; use sha2::{Digest, Sha256}; -use sqlx::{PgPool, Row}; +use sqlx::{Executor, PgPool, Postgres, Row, Transaction}; use uuid::Uuid; use buzz_core::CommunityId; @@ -310,6 +316,8 @@ pub async fn create_workflow( } /// Insert or update a workflow at the caller-supplied NIP-33 `d`-tag UUID. +/// Runtime eligibility is projected from the same canonical definition on every +/// save; an omitted `enabled` field keeps the workflow language's true default. /// /// Updates are allowed only when the existing row has the same owner and /// channel. That keeps a learned workflow UUID from becoming a cross-user or @@ -324,16 +332,66 @@ pub async fn upsert_workflow( name: &str, definition_json: &str, definition_hash: &[u8], +) -> Result<()> { + upsert_workflow_on( + pool, + community_id, + id, + channel_id, + owner_pubkey, + name, + definition_json, + definition_hash, + ) + .await +} + +/// Upsert the runtime projection in the same transaction as its signed definition. +#[allow(clippy::too_many_arguments)] +pub async fn upsert_workflow_in_transaction( + tx: &mut Transaction<'_, Postgres>, + community_id: CommunityId, + id: Uuid, + channel_id: Option, + owner_pubkey: &[u8], + name: &str, + definition_json: &str, + definition_hash: &[u8], +) -> Result<()> { + upsert_workflow_on( + &mut **tx, + community_id, + id, + channel_id, + owner_pubkey, + name, + definition_json, + definition_hash, + ) + .await +} + +#[allow(clippy::too_many_arguments)] +async fn upsert_workflow_on<'e>( + executor: impl Executor<'e, Database = Postgres>, + community_id: CommunityId, + id: Uuid, + channel_id: Option, + owner_pubkey: &[u8], + name: &str, + definition_json: &str, + definition_hash: &[u8], ) -> Result<()> { let row = sqlx::query( r#" INSERT INTO workflows (community_id, id, name, owner_pubkey, channel_id, definition, definition_hash, status, enabled) - VALUES ($1, $2, $3, $4, $5, $6::jsonb, $7, 'active', TRUE) + VALUES ($1, $2, $3, $4, $5, $6::jsonb, $7, 'active', COALESCE(($6::jsonb->>'enabled')::boolean, TRUE)) ON CONFLICT (community_id, id) DO UPDATE SET name = EXCLUDED.name, definition = EXCLUDED.definition, definition_hash = EXCLUDED.definition_hash, + enabled = EXCLUDED.enabled, updated_at = NOW() WHERE workflows.owner_pubkey = EXCLUDED.owner_pubkey AND workflows.channel_id IS NOT DISTINCT FROM EXCLUDED.channel_id @@ -347,7 +405,7 @@ pub async fn upsert_workflow( .bind(channel_id) .bind(definition_json) .bind(definition_hash) - .fetch_optional(pool) + .fetch_optional(executor) .await?; if row.is_none() { @@ -369,6 +427,23 @@ pub async fn get_workflow( pool: &PgPool, community_id: CommunityId, id: Uuid, +) -> Result { + get_workflow_on(pool, community_id, id).await +} + +/// Read workflow state after taking its coordinate lock, without leaving the event transaction. +pub async fn get_workflow_in_transaction( + tx: &mut Transaction<'_, Postgres>, + community_id: CommunityId, + id: Uuid, +) -> Result { + get_workflow_on(&mut **tx, community_id, id).await +} + +async fn get_workflow_on<'e>( + executor: impl Executor<'e, Database = Postgres>, + community_id: CommunityId, + id: Uuid, ) -> Result { let row = sqlx::query( r#" @@ -380,7 +455,7 @@ pub async fn get_workflow( ) .bind(community_id.as_uuid()) .bind(id) - .fetch_optional(pool) + .fetch_optional(executor) .await? .ok_or_else(|| DbError::NotFound(format!("workflow {id}")))?; @@ -795,38 +870,6 @@ pub async fn delete_workflow_for_owner( // -- Workflow Run CRUD -------------------------------------------------------- -/// Insert a new workflow run. Returns the new run's UUID. -/// -/// `trigger_context` is the serialized `TriggerContext` for this run. It is stored -/// so that post-approval resume steps can restore the original trigger data and -/// correctly resolve `{{trigger.*}}` template variables. -pub async fn create_workflow_run( - pool: &PgPool, - community_id: CommunityId, - workflow_id: Uuid, - trigger_event_id: Option<&[u8]>, - trigger_context: Option<&serde_json::Value>, -) -> Result { - let id = Uuid::new_v4(); - - sqlx::query( - r#" - INSERT INTO workflow_runs - (community_id, id, workflow_id, status, trigger_event_id, current_step, execution_trace, trigger_context) - VALUES ($1, $2, $3, 'pending', $4, 0, '[]', $5) - "#, - ) - .bind(community_id.as_uuid()) - .bind(id) - .bind(workflow_id) - .bind(trigger_event_id) - .bind(trigger_context) - .execute(pool) - .await?; - - Ok(id) -} - /// Fetch a single workflow run by ID, scoped to its community. pub async fn get_workflow_run( pool: &PgPool, @@ -1271,19 +1314,18 @@ pub async fn find_by_owner_and_name( // -- Run and approval Db API -------------------------------------------------- impl Db { - /// Create a new workflow run. + /// Admit a run only for the still-current, active/enabled selected workflow. + /// See [`create_workflow_run`] for the atomic admission contract. #[datastore_span(name = "create_workflow_run", system = "postgresql")] pub async fn create_workflow_run( &self, - community_id: CommunityId, - workflow_id: Uuid, + workflow: &WorkflowRecord, trigger_event_id: Option<&[u8]>, trigger_context: Option<&serde_json::Value>, - ) -> Result { + ) -> Result> { crate::workflow::create_workflow_run( &self.pool, - community_id, - workflow_id, + workflow, trigger_event_id, trigger_context, ) @@ -2380,9 +2422,11 @@ mod postgres_tests { .expect("claim wins"); // Create the run the won claim is responsible for, then attach it. - let run_id = create_workflow_run(&pool, community, workflow_id, None, None) + let workflow = get_workflow(&pool, community, workflow_id).await.unwrap(); + let run_id = create_workflow_run(&pool, &workflow, None, None) .await - .expect("create run ok"); + .expect("create run ok") + .expect("current workflow admitted"); let attached = attach_scheduled_workflow_run(&pool, community, workflow_id, scheduled_for, run_id) @@ -2409,9 +2453,10 @@ mod postgres_tests { // A second attach is a no-op: the `workflow_run_id IS NULL` guard means // an already-linked claim is never re-pointed to a different run. - let other_run = create_workflow_run(&pool, community, workflow_id, None, None) + let other_run = create_workflow_run(&pool, &workflow, None, None) .await - .expect("create second run ok"); + .expect("create second run ok") + .expect("current workflow admitted"); let reattached = attach_scheduled_workflow_run(&pool, community, workflow_id, scheduled_for, other_run) .await @@ -2690,12 +2735,16 @@ mod postgres_tests { insert_workflow_with_ids(&pool, community_a, workflow_id, channel_id, "wf-A").await; insert_workflow_with_ids(&pool, community_b, workflow_id, Uuid::new_v4(), "wf-B").await; - let run_a = create_workflow_run(&pool, community_a, workflow_id, None, None) + let workflow_a = get_workflow(&pool, community_a, workflow_id).await.unwrap(); + let workflow_b = get_workflow(&pool, community_b, workflow_id).await.unwrap(); + let run_a = create_workflow_run(&pool, &workflow_a, None, None) .await - .expect("run A"); - let run_b = create_workflow_run(&pool, community_b, workflow_id, None, None) + .expect("run A") + .expect("current workflow admitted"); + let run_b = create_workflow_run(&pool, &workflow_b, None, None) .await - .expect("run B"); + .expect("run B") + .expect("current workflow admitted"); let token = "shared-approval-token"; let expires = Utc::now() + chrono::Duration::hours(1); diff --git a/crates/buzz-db/src/store/workflow/lifecycle.rs b/crates/buzz-db/src/store/workflow/lifecycle.rs new file mode 100644 index 00000000000..1c466eb5804 --- /dev/null +++ b/crates/buzz-db/src/store/workflow/lifecycle.rs @@ -0,0 +1,166 @@ +//! Forward-only workflow deletion proofs, serialized with NIP-33 replacement. +//! One fixed-size row per deleted coordinate; no workflow content or secrets. + +use buzz_core::{kind::KIND_WORKFLOW_DEF, CommunityId}; +use chrono::{DateTime, Utc}; +use sqlx::{Postgres, Transaction}; +use uuid::Uuid; + +use crate::{observability, replaceable::event_replacement_lock_key, Result}; + +/// Signed live head and its server-resolved channel. +#[derive(Debug, sqlx::FromRow)] +pub struct Head { + /// Signed timestamp, not server receipt time. + pub created_at: DateTime, + /// Exact signed definition ID. + pub id: Vec, + /// Stored channel scope; absent legacy scope is not authoritative. + pub channel_id: Option, +} + +/// Latest committed forward deletion for this coordinate. +#[derive(Debug, sqlx::FromRow)] +pub struct Deletion { + /// Actual channel resolved by the successful transaction. + pub channel_id: Uuid, + /// Older/equal unseen saves must not revive this coordinate. + pub deleted_through: DateTime, + /// Exact deletion whose atomic transaction committed. + pub event_id: Vec, +} + +/// Take the same lock as canonical kind-30620 replacement. The caller owns the transaction. +pub async fn lock_coordinate( + tx: &mut Transaction<'_, Postgres>, + community: CommunityId, + owner: &[u8], + id: Uuid, +) -> Result<()> { + let d_tag = id.to_string(); + let key = event_replacement_lock_key( + community, + KIND_WORKFLOW_DEF as i32, + owner, + Some(d_tag.as_bytes()), + ); + observability::observe_advisory_lock( + observability::LockType::Replacement, + sqlx::query("SELECT pg_advisory_xact_lock($1)") + .bind(key) + .execute(&mut **tx), + ) + .await?; + Ok(()) +} + +/// Read the live definition under the coordinate lock. +pub async fn head( + tx: &mut Transaction<'_, Postgres>, + community: CommunityId, + owner: &[u8], + id: Uuid, +) -> Result> { + Ok(sqlx::query_as( + "SELECT created_at, id, channel_id 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, id ASC LIMIT 1", + ).bind(community.as_uuid()).bind(KIND_WORKFLOW_DEF as i32).bind(owner).bind(id.to_string()) + .fetch_optional(&mut **tx).await?) +} + +/// Read the trusted forward cutoff. Accepted legacy deletion events are never consulted. +pub async fn deletion( + tx: &mut Transaction<'_, Postgres>, + community: CommunityId, + owner: &[u8], + id: Uuid, +) -> Result> { + Ok(sqlx::query_as( + "SELECT channel_id, deleted_through, event_id FROM workflow_deletions \ + WHERE community_id=$1 AND owner_pubkey=$2 AND workflow_id=$3", + ) + .bind(community.as_uuid()) + .bind(owner) + .bind(id) + .fetch_optional(&mut **tx) + .await?) +} + +/// Recognize exact stored events, including soft-deleted versions, without reapplying them. +pub async fn event_seen( + tx: &mut Transaction<'_, Postgres>, + community: CommunityId, + event: &nostr::Event, +) -> Result { + Ok(sqlx::query_scalar( + "SELECT EXISTS(SELECT 1 FROM events WHERE community_id=$1 AND id=$2 AND created_at=$3)", + ) + .bind(community.as_uuid()) + .bind(event.id.as_bytes().as_slice()) + .bind( + DateTime::from_timestamp(event.created_at.as_secs() as i64, 0).ok_or( + crate::DbError::InvalidTimestamp(event.created_at.as_secs() as i64), + )?, + ) + .fetch_one(&mut **tx) + .await?) +} + +/// Mutate both projections and advance the proof in the caller's transaction. +/// The caller must hold the coordinate lock, validate owner/channel and timestamp, +/// and insert the deletion event in this same transaction before committing. +pub async fn delete_in_transaction( + tx: &mut Transaction<'_, Postgres>, + community: CommunityId, + owner: &[u8], + id: Uuid, + channel: Uuid, + event: &nostr::Event, +) -> Result<()> { + let through = DateTime::from_timestamp(event.created_at.as_secs() as i64, 0).ok_or( + crate::DbError::InvalidTimestamp(event.created_at.as_secs() as i64), + )?; + sqlx::query("DELETE FROM workflows WHERE community_id=$1 AND id=$2 AND owner_pubkey=$3 AND channel_id=$4") + .bind(community.as_uuid()).bind(id).bind(owner).bind(channel).execute(&mut **tx).await?; + sqlx::query( + "UPDATE events SET deleted_at=NOW() WHERE community_id=$1 AND kind=$2 AND pubkey=$3 \ + AND d_tag=$4 AND channel_id=$5 AND created_at <= $6 AND deleted_at IS NULL", + ) + .bind(community.as_uuid()) + .bind(KIND_WORKFLOW_DEF as i32) + .bind(owner) + .bind(id.to_string()) + .bind(channel) + .bind(through) + .execute(&mut **tx) + .await?; + sqlx::query("INSERT INTO workflow_deletions (community_id, owner_pubkey, workflow_id, channel_id, deleted_through, event_id) \ + VALUES ($1,$2,$3,$4,$5,$6) ON CONFLICT (community_id, owner_pubkey, workflow_id) \ + DO UPDATE SET deleted_through=EXCLUDED.deleted_through, event_id=EXCLUDED.event_id \ + WHERE workflow_deletions.deleted_through < EXCLUDED.deleted_through") + .bind(community.as_uuid()).bind(owner).bind(id).bind(channel).bind(through) + .bind(event.id.as_bytes().as_slice()).execute(&mut **tx).await?; + Ok(()) +} + +/// Lock the active member and channel rows until the lifecycle transaction ends. +/// Returns the actual role; open-channel visibility is not workflow-author authority. +pub async fn channel_role( + tx: &mut Transaction<'_, Postgres>, + community: CommunityId, + channel: Uuid, + owner: &[u8], +) -> Result> { + Ok(sqlx::query_scalar( + "SELECT cm.role::text FROM channel_members cm JOIN channels c \ + ON c.community_id=cm.community_id AND c.id=cm.channel_id \ + WHERE cm.community_id=$1 AND cm.channel_id=$2 AND cm.pubkey=$3 \ + AND cm.removed_at IS NULL AND c.deleted_at IS NULL AND c.archived_at IS NULL \ + FOR SHARE OF c, cm", + ) + .bind(community.as_uuid()) + .bind(channel) + .bind(owner) + .fetch_optional(&mut **tx) + .await?) +} diff --git a/crates/buzz-db/src/store/workflow/run_admission.rs b/crates/buzz-db/src/store/workflow/run_admission.rs new file mode 100644 index 00000000000..9777c30180f --- /dev/null +++ b/crates/buzz-db/src/store/workflow/run_admission.rs @@ -0,0 +1,251 @@ +//! Atomic admission of new work against the selected workflow incarnation. + +use sqlx::PgPool; +use uuid::Uuid; + +use super::WorkflowRecord; +use crate::Result; + +/// Insert a pending run only if the selected workflow is still current and eligible. +/// +/// The record must be the one used to select/authorize the execution definition; +/// its community is server-resolved provenance, never a client-supplied tenant. +/// Returns `None` if the workflow is missing, inactive, disabled, or no longer +/// matches the selected definition hash, owner, channel or creation incarnation. +/// Database failures remain errors. Callers must not execute without `Some(id)`. +/// +/// A SHARE row lock conflicts with updates and deletion until the INSERT commits +/// (unlike the FK's KEY SHARE lock). PostgreSQL rechecks the predicates after a +/// concurrent writer settles; selection and insertion have no unlocked gap. +/// Already admitted runs are intentionally unaffected by this boundary. +/// +/// `trigger_context` stores the serialized `TriggerContext` so approval resume +/// can restore the original trigger data and `{{trigger.*}}` template variables. +pub async fn create_workflow_run( + pool: &PgPool, + workflow: &WorkflowRecord, + trigger_event_id: Option<&[u8]>, + trigger_context: Option<&serde_json::Value>, +) -> Result> { + Ok(sqlx::query_scalar( + r#" + WITH eligible AS ( + SELECT community_id, id FROM workflows + WHERE community_id = $1 AND id = $2 + AND status = 'active' AND enabled = TRUE + AND definition_hash = $3 AND owner_pubkey = $4 + AND channel_id IS NOT DISTINCT FROM $5 AND created_at = $6 + FOR SHARE + ) + INSERT INTO workflow_runs + (community_id, id, workflow_id, status, trigger_event_id, current_step, execution_trace, trigger_context) + SELECT community_id, $7, id, 'pending', $8, 0, '[]', $9 FROM eligible + RETURNING id + "#, + ) + .bind(workflow.community_id.as_uuid()) + .bind(workflow.id) + .bind(&workflow.definition_hash) + .bind(&workflow.owner_pubkey) + .bind(workflow.channel_id) + .bind(workflow.created_at) + .bind(Uuid::new_v4()) + .bind(trigger_event_id) + .bind(trigger_context) + .fetch_optional(pool) + .await?) +} + +#[cfg(test)] +mod postgres_tests { + use super::*; + use crate::{workflow, CommunityId}; + use std::time::Duration; + + async fn fixture() -> (PgPool, WorkflowRecord) { + let pool = PgPool::connect(&crate::test_support::database_url()) + .await + .expect("test database"); + let community = CommunityId::from_uuid(Uuid::new_v4()); + sqlx::query("INSERT INTO communities (id, host) VALUES ($1, $2)") + .bind(community.as_uuid()) + .bind(format!("admission-{}.example", community.as_uuid())) + .execute(&pool) + .await + .unwrap(); + let owner = [0x31; 32]; + crate::user::ensure_user(&pool, community, &owner) + .await + .unwrap(); + let id = workflow::create_workflow( + &pool, + community, + None, + &owner, + "admission", + r#"{"enabled":true}"#, + &[0x41; 32], + ) + .await + .unwrap(); + let selected = workflow::get_workflow(&pool, community, id).await.unwrap(); + (pool, selected) + } + + #[tokio::test] + #[ignore = "requires Postgres"] + async fn admission_requires_selected_identity_revision_and_eligibility() { + let (pool, selected) = fixture().await; + let db = crate::Db::from_pool(pool.clone()); + let trigger = serde_json::json!({"text":"original trigger"}); + let event = [0x51; 32]; + let id = db + .create_workflow_run(&selected, Some(&event), Some(&trigger)) + .await + .unwrap() + .expect("current record admitted"); + let run = db + .get_workflow_run(selected.community_id, id) + .await + .unwrap(); + assert_eq!(run.workflow_id, selected.id); + assert_eq!(run.community_id, selected.community_id); + assert_eq!(run.trigger_event_id.as_deref(), Some(event.as_slice())); + assert_eq!(run.trigger_context, Some(trigger)); + assert_eq!(run.status, workflow::RunStatus::Pending); + + for field in ["community", "id", "owner", "channel", "hash", "incarnation"] { + let mut stale = selected.clone(); + match field { + "community" => stale.community_id = CommunityId::from_uuid(Uuid::new_v4()), + "id" => stale.id = Uuid::new_v4(), + "owner" => stale.owner_pubkey = vec![0x32; 32], + "channel" => stale.channel_id = Some(Uuid::new_v4()), + "hash" => stale.definition_hash = vec![0x42; 32], + "incarnation" => stale.created_at += chrono::Duration::microseconds(1), + _ => unreachable!(), + } + assert!( + db.create_workflow_run(&stale, None, None) + .await + .unwrap() + .is_none(), + "{field}" + ); + } + workflow::set_workflow_enabled(&pool, selected.community_id, selected.id, false) + .await + .unwrap(); + assert!(db + .create_workflow_run(&selected, None, None) + .await + .unwrap() + .is_none()); + workflow::set_workflow_enabled(&pool, selected.community_id, selected.id, true) + .await + .unwrap(); + for status in [ + workflow::WorkflowStatus::Disabled, + workflow::WorkflowStatus::Archived, + ] { + workflow::update_workflow_status(&pool, selected.community_id, selected.id, status) + .await + .unwrap(); + assert!(db + .create_workflow_run(&selected, None, None) + .await + .unwrap() + .is_none()); + } + // Rejected attempts left no run rows, and did not alter an admitted run. + assert_eq!( + db.list_workflow_runs(selected.community_id, selected.id, 100) + .await + .unwrap() + .len(), + 1 + ); + workflow::delete_workflow(&pool, selected.community_id, selected.id) + .await + .unwrap(); + assert!(db + .create_workflow_run(&selected, None, None) + .await + .unwrap() + .is_none()); + } + + /// Observe a real server lock wait, not a sleep that merely hopes admission ran. + async fn wait_for_admission_lock( + pool: &PgPool, + writer_pid: i32, + admission: &tokio::task::JoinHandle>>, + ) { + tokio::time::timeout(Duration::from_secs(5), async { + loop { + assert!(!admission.is_finished(), "admission bypassed the in-flight workflow writer"); + let blocked: bool = sqlx::query_scalar( + "SELECT EXISTS(SELECT 1 FROM pg_stat_activity WHERE datname = current_database() \ + AND query LIKE '%WITH eligible AS%' AND $1 = ANY(pg_blocking_pids(pid)))", + ).bind(writer_pid).fetch_one(pool).await.unwrap(); + if blocked { return; } + tokio::time::sleep(Duration::from_millis(10)).await; + } + }).await.expect("admission must wait for the workflow row lock"); + } + + #[tokio::test] + #[ignore = "requires Postgres"] + async fn cluster_global_admission_waits_for_update_and_disable_then_rechecks() { + let (pool, selected) = fixture().await; + for mutation in [ + "UPDATE workflows SET definition_hash = $3 WHERE community_id = $1 AND id = $2", + "UPDATE workflows SET enabled = FALSE WHERE community_id = $1 AND id = $2 AND definition_hash <> $3", + ] { + let current = workflow::get_workflow(&pool, selected.community_id, selected.id).await.unwrap(); + let mut tx = pool.begin().await.unwrap(); + let pid: i32 = sqlx::query_scalar("SELECT pg_backend_pid()").fetch_one(&mut *tx).await.unwrap(); + sqlx::query(mutation).bind(selected.community_id.as_uuid()).bind(selected.id).bind(vec![0x42u8; 32]) + .execute(&mut *tx).await.unwrap(); + let db = crate::Db::from_pool(pool.clone()); + let admission = tokio::spawn(async move { db.create_workflow_run(¤t, None, None).await }); + wait_for_admission_lock(&pool, pid, &admission).await; + tx.commit().await.unwrap(); + assert!(tokio::time::timeout(Duration::from_secs(5), admission).await.unwrap().unwrap().unwrap().is_none()); + // Restore eligibility for the next independent in-flight mutation. + sqlx::query("UPDATE workflows SET definition_hash = $3, enabled = TRUE WHERE community_id = $1 AND id = $2") + .bind(selected.community_id.as_uuid()).bind(selected.id).bind(&selected.definition_hash) + .execute(&pool).await.unwrap(); + } + assert!( + workflow::list_workflow_runs(&pool, selected.community_id, selected.id, 100) + .await + .unwrap() + .is_empty() + ); + + // A rolled-back disable must release the waiter and admit the unchanged record. + let mut tx = pool.begin().await.unwrap(); + let pid: i32 = sqlx::query_scalar("SELECT pg_backend_pid()") + .fetch_one(&mut *tx) + .await + .unwrap(); + sqlx::query("UPDATE workflows SET enabled = FALSE WHERE community_id = $1 AND id = $2") + .bind(selected.community_id.as_uuid()) + .bind(selected.id) + .execute(&mut *tx) + .await + .unwrap(); + let db = crate::Db::from_pool(pool.clone()); + let admission = + tokio::spawn(async move { db.create_workflow_run(&selected, None, None).await }); + wait_for_admission_lock(&pool, pid, &admission).await; + tx.rollback().await.unwrap(); + assert!(tokio::time::timeout(Duration::from_secs(5), admission) + .await + .unwrap() + .unwrap() + .unwrap() + .is_some()); + } +} diff --git a/crates/buzz-relay/src/api/bridge.rs b/crates/buzz-relay/src/api/bridge.rs index 37c549610de..664bf7ed826 100644 --- a/crates/buzz-relay/src/api/bridge.rs +++ b/crates/buzz-relay/src/api/bridge.rs @@ -2180,9 +2180,10 @@ pub async fn workflow_webhook( let run_id = state .db - .create_workflow_run(community_id, id, None, trigger_ctx_json.as_ref()) + .create_workflow_run(&workflow, None, trigger_ctx_json.as_ref()) .await - .map_err(|e| super::internal_error(&format!("db error: {e}")))?; + .map_err(|e| super::internal_error(&format!("db error: {e}")))? + .ok_or_else(|| not_found("workflow not found"))?; // Spawn workflow execution asynchronously. let engine = Arc::clone(&state.workflow_engine); diff --git a/crates/buzz-relay/src/config.rs b/crates/buzz-relay/src/config.rs index e035752ec3a..5f332d2ca08 100644 --- a/crates/buzz-relay/src/config.rs +++ b/crates/buzz-relay/src/config.rs @@ -209,6 +209,13 @@ pub struct Config { /// are permitted regardless of auth method (API token, NIP-42). pub require_relay_membership: bool, + /// Advertise the host-wide workflow lifecycle contract in NIP-11. + /// Default off. Enable `BUZZ_ADVERTISE_WORKFLOW_LIFECYCLE=true` only after + /// every endpoint and existing connection for every served host runs + /// compatible lifecycle code; see deploy/charts/buzz/README.md#workflow-lifecycle-activation. + /// This gates discovery only, never the atomic save/delete enforcement. + pub advertise_workflow_lifecycle: bool, + /// Whether this deployment can serve huddle (voice) audio. /// /// Huddle audio frames are relayed peer-to-peer *within a single pod* @@ -671,6 +678,10 @@ impl Config { .map(|v| v == "true" || v == "1") .unwrap_or(false); + let advertise_workflow_lifecycle = std::env::var("BUZZ_ADVERTISE_WORKFLOW_LIFECYCLE") + .map(|v| v == "true" || v == "1") + .unwrap_or(false); + // Defaults true → single-pod (N=1) keeps today's huddle behavior. A // horizontally-scaled deployment sets this false; see the field doc. let huddle_audio_available = std::env::var("BUZZ_HUDDLE_AUDIO_AVAILABLE") @@ -1226,6 +1237,7 @@ impl Config { metrics_port, pubkey_allowlist_enabled, require_relay_membership, + advertise_workflow_lifecycle, huddle_audio_available, mesh, mesh_demo_echo, @@ -1334,6 +1346,42 @@ mod tests { assert!(found.is_empty(), "unrelated vars must not warn: {found:?}"); } + #[test] + fn workflow_lifecycle_advertisement_requires_explicit_opt_in() { + // Config::from_env readers elsewhere do not share ENV_MUTEX. Keep + // this flag matrix in an isolated child rather than racing them. + const CHILD: &str = "BUZZ_TEST_WORKFLOW_ADVERTISEMENT_CONFIG_CHILD"; + let _guard = ENV_MUTEX.lock().unwrap(); + if std::env::var_os(CHILD).is_none() { + crate::test_support::run_exact_test_child( + "config::tests::workflow_lifecycle_advertisement_requires_explicit_opt_in", + CHILD, + ); + return; + } + for (value, enabled) in [ + (None, false), + (Some(""), false), + (Some("false"), false), + (Some("0"), false), + (Some("typo"), false), + (Some("true"), true), + (Some("1"), true), + ] { + match value { + Some(value) => std::env::set_var("BUZZ_ADVERTISE_WORKFLOW_LIFECYCLE", value), + None => std::env::remove_var("BUZZ_ADVERTISE_WORKFLOW_LIFECYCLE"), + } + assert_eq!( + Config::from_env() + .expect("config") + .advertise_workflow_lifecycle, + enabled, + "advertisement config {value:?}" + ); + } + } + #[test] fn defaults_are_valid() { let _guard = ENV_MUTEX.lock().unwrap(); diff --git a/crates/buzz-relay/src/handlers/command_executor.rs b/crates/buzz-relay/src/handlers/command_executor.rs index 074f6b391d0..96d2c8e3572 100644 --- a/crates/buzz-relay/src/handlers/command_executor.rs +++ b/crates/buzz-relay/src/handlers/command_executor.rs @@ -92,12 +92,13 @@ enum PersistResult { /// If the event is a duplicate (ON CONFLICT DO NOTHING), the transaction is /// rolled back and `PersistResult::Duplicate` is returned — no mutations needed. /// -/// NOTE: Domain mutations (open_dm, upsert_workflow, etc.) execute on the +/// NOTE: Other domain mutations (open_dm, etc.) execute on the /// connection pool, NOT inside this transaction. The pattern is idempotent but /// not strictly atomic: if a mutation succeeds but commit fails, the mutation /// persists without the event record. On retry, the event INSERT succeeds /// (no conflict), and the mutation re-executes — which is safe for idempotent -/// operations (open_dm, hide_dm, update_approval, upsert_workflow). +/// operations (open_dm, hide_dm, update_approval). Workflow definitions instead +/// compose their runtime projection in this transaction. #[datastore_span(name = "persist_command_event", system = "postgresql")] async fn persist_command_event( db: &buzz_db::Db, @@ -130,6 +131,53 @@ async fn persist_command_event( } let kind = event.kind.as_u16() as i32; + if kind == KIND_WORKFLOW_DEF as i32 { + use buzz_db::workflow::lifecycle; + let id = Uuid::parse_str(d_tag) + .map_err(|_| IngestError::Rejected("invalid: bad workflow_id format".into()))?; + // UUID aliases otherwise share a runtime row but not a replacement lock. + if d_tag != id.to_string() + || event + .tags + .iter() + .filter(|tag| tag.kind().to_string() == "d") + .count() + != 1 + { + return Err(IngestError::Rejected( + "invalid: workflow d tag must be one canonical UUID".into(), + )); + } + let owner = event.pubkey.to_bytes(); + lifecycle::lock_coordinate(&mut tx, tenant.community(), &owner, id) + .await + .map_err(|e| { + IngestError::Internal(format!("error: workflow coordinate lock: {e}")) + })?; + if lifecycle::event_seen(&mut tx, tenant.community(), event) + .await + .map_err(|e| IngestError::Internal(format!("error: workflow replay lookup: {e}")))? + { + return Ok(PersistResult::Duplicate); + } + if let Some(deletion) = lifecycle::deletion(&mut tx, tenant.community(), &owner, id) + .await + .map_err(|e| { + IngestError::Internal(format!("error: workflow deletion lookup: {e}")) + })? + { + if event.created_at.as_secs() as i64 <= deletion.deleted_through.timestamp() { + return Err(IngestError::Rejected( + "conflict: workflow save predates a committed deletion".into(), + )); + } + if channel_id != Some(deletion.channel_id) { + return Err(IngestError::Rejected( + "forbidden: workflow belongs to a different channel".into(), + )); + } + } + } let (expected_revision, revision_error) = match parse_expected_workflow_revision( kind, extract_tag(event, "expected-revision").as_deref(), @@ -657,44 +705,65 @@ async fn handle_workflow_def( let workflow_id = Uuid::parse_str(&workflow_id_str) .map_err(|_| IngestError::Rejected("invalid: bad workflow_id format".into()))?; - // 2. Validate caller has channel access (minimum: is a member) - let is_member = state - .is_member_cached(tenant.community(), channel_id, &self_bytes) - .await - .map_err(|e| IngestError::Internal(format!("error: membership check: {e}")))?; - if !is_member { - return Err(IngestError::Rejected( - "forbidden: not a member of this channel".into(), - )); - } - // 3. Parse YAML from event.content let (def, definition_json_str) = buzz_workflow::WorkflowEngine::parse_yaml(&event.content) .map_err(|e| IngestError::Rejected(format!("invalid: workflow YAML parse error: {e}")))?; let workflow_name = extract_tag(event, "name").unwrap_or_else(|| def.name.clone()); - // SEC-006: definitions with exfiltration-capable actions (call_webhook) - // require elevated channel authority to save — plain membership is not - // enough, because the workflow will forward channel content outward with - // the owner's standing authority. Fail-closed on lookup errors. - if def.requires_elevated_authority() { - let role = state - .db - .get_member_role(tenant.community(), channel_id, &self_bytes) - .await - .map_err(|e| IngestError::Internal(format!("error: role check: {e}")))?; - if !matches!(role.as_deref(), Some("owner") | Some("admin")) { - return Err(IngestError::Rejected( - "forbidden: workflows with call_webhook actions require the owner or admin role" - .into(), - )); - } - } - let mut definition_json: serde_json::Value = serde_json::from_str(&definition_json_str) .map_err(|e| IngestError::Internal(format!("error: json parse of definition: {e}")))?; - let existing_workflow = match state.db.get_workflow(tenant.community(), workflow_id).await { + let channel_tags: Vec<_> = event + .tags + .iter() + .filter(|tag| tag.kind().to_string() == "h") + .collect(); + if channel_tags.len() != 1 + || channel_tags[0].as_slice().len() != 2 + || channel_id_str != channel_id.to_string() + { + return Err(IngestError::Rejected( + "invalid: workflow h tag must be one canonical channel UUID".into(), + )); + } + super::ingest::check_token_channel_access(auth, channel_id).map_err(IngestError::AuthFailed)?; + + // Persist the command event — returns open transaction + let mut tx = match persist_command_event(&state.db, tenant, event, None).await? { + PersistResult::Duplicate => { + return Ok(IngestResult { + event_id: event.id.to_hex(), + accepted: true, + message: "duplicate: already processed".into(), + }); + } + PersistResult::Inserted(tx) => tx, + }; + + let role = buzz_db::workflow::lifecycle::channel_role( + &mut tx, + tenant.community(), + channel_id, + &self_bytes, + ) + .await + .map_err(|e| IngestError::Internal(format!("error: workflow channel authority: {e}")))? + .ok_or_else(|| { + IngestError::Rejected("forbidden: workflow requires active channel membership".into()) + })?; + if def.requires_elevated_authority() && !matches!(role.as_str(), "owner" | "admin") { + return Err(IngestError::Rejected( + "forbidden: workflows with call_webhook actions require the owner or admin role".into(), + )); + } + + let existing_workflow = match buzz_db::workflow::get_workflow_in_transaction( + &mut tx, + tenant.community(), + workflow_id, + ) + .await + { Ok(workflow) => { if workflow.owner_pubkey != self_bytes || workflow.channel_id != Some(channel_id) { return Err(IngestError::Rejected( @@ -737,66 +806,37 @@ async fn handle_workflow_def( .map_err(|e| IngestError::Internal(format!("error: json serialize: {e}")))?; let hash = compute_definition_hash(&definition_json_final); - // Persist the command event — returns open transaction - let tx = match persist_command_event(&state.db, tenant, event, None).await? { - PersistResult::Duplicate => { - return Ok(IngestResult { - event_id: event.id.to_hex(), - accepted: true, - message: "duplicate: already processed".into(), - }); - } - PersistResult::Inserted(tx) => tx, - }; - - // 4. Execute: upsert by the NIP-33 d-tag UUID. A retry updates the same - // row instead of creating another enabled workflow that would fan out on - // every matching event. The workflow's community is the request's - // server-bound tenant — never re-derived from the (client-supplied) channel - // id. `community_of_channel(channel_id)` is ambiguous when the same channel - // UUID exists in two communities and could mint the workflow under the wrong - // tenant; `tenant.community()` is the authoritative owner. We then verify the - // channel actually exists *inside that community* (scoped `get_channel`), - // which fails closed if the client named a channel that belongs to a - // different community — the same guarantee the `(community_id, channel_id)` - // composite FK enforces on insert, surfaced here as a clean rejection. + // The tenant is host-bound; channel_role verified and locked this channel in it. let community_id = tenant.community(); - state - .db - .get_channel_for_event_write(community_id, channel_id) - .await - .map_err(|_| IngestError::Rejected("invalid: workflow channel not found".into()))?; + buzz_db::workflow::upsert_workflow_in_transaction( + &mut tx, + community_id, + workflow_id, + Some(channel_id), + &self_bytes, + &workflow_name, + &definition_json_final, + &hash, + ) + .await + .map_err(|e| match e { + DbError::AccessDenied(_) => IngestError::Rejected( + "forbidden: workflow belongs to a different owner or channel".into(), + ), + other => IngestError::Internal(format!("error: db upsert_workflow: {other}")), + })?; - state - .db - .upsert_workflow( - community_id, - workflow_id, - Some(channel_id), - &self_bytes, - &workflow_name, - &definition_json_final, - &hash, - ) + // Both signed intent and runtime projection become visible together. + tx.commit() .await - .map_err(|e| match e { - DbError::AccessDenied(_) => IngestError::Rejected( - "forbidden: workflow belongs to a different owner or channel".into(), - ), - other => IngestError::Internal(format!("error: db upsert_workflow: {other}")), - })?; + .map_err(|e| IngestError::Internal(format!("error: commit transaction: {e}")))?; - // Drop the trigger-path cache entry so the new/updated definition fires on - // the next matching event instead of after the cache TTL. + // Invalidate after commit; an older in-flight cache fill can still race this + // eviction (the engine bounds that existing race by its cache TTL). state .workflow_engine .invalidate_channel_workflows(community_id, channel_id); - // Commit the event transaction after the idempotent workflow upsert succeeds. - tx.commit() - .await - .map_err(|e| IngestError::Internal(format!("error: commit transaction: {e}")))?; - // 5. Return response let mut resp = serde_json::json!({ "workflow_id": workflow_id.to_string(), @@ -913,14 +953,12 @@ async fn handle_workflow_trigger( let event_id_bytes = event.id.as_bytes().to_vec(); let run_id = state .db - .create_workflow_run( - community_id, - workflow_id, - Some(&event_id_bytes), - trigger_ctx_json.as_ref(), - ) + .create_workflow_run(&workflow, Some(&event_id_bytes), trigger_ctx_json.as_ref()) .await - .map_err(|e| IngestError::Internal(format!("error: db create_workflow_run: {e}")))?; + .map_err(|e| IngestError::Internal(format!("error: db create_workflow_run: {e}")))? + .ok_or_else(|| { + IngestError::Rejected("forbidden: workflow changed or is no longer active".into()) + })?; // Finalize the idempotency record after the separate run creation succeeds. tx.commit() @@ -1635,3 +1673,6 @@ mod postgres_tests { )); } } + +#[cfg(test)] +mod run_admission_postgres_tests; diff --git a/crates/buzz-relay/src/handlers/command_executor/run_admission_postgres_tests.rs b/crates/buzz-relay/src/handlers/command_executor/run_admission_postgres_tests.rs new file mode 100644 index 00000000000..7fd09d08bb1 --- /dev/null +++ b/crates/buzz-relay/src/handlers/command_executor/run_admission_postgres_tests.rs @@ -0,0 +1,259 @@ +//! Production manual/webhook admission after a concurrent definition update. +use super::*; +use axum::{ + extract::{Path, Query, State}, + http::{HeaderMap, StatusCode}, +}; +use buzz_core::channel::{ChannelType, ChannelVisibility}; +use nostr::{EventBuilder, Keys, Kind, Tag}; +use std::{future::Future, pin::Pin, sync::Mutex, time::Duration}; + +#[derive(Default)] +struct Sink(Mutex>); +impl buzz_workflow::ActionSink for Sink { + fn send_message( + &self, + _: CommunityId, + _: &str, + text: &str, + _: &str, + _: &str, + _: Option<&str>, + ) -> Pin> + Send + '_>> + { + self.0.lock().unwrap().push(text.into()); + Box::pin(async { Ok("ab".repeat(32)) }) + } +} + +async fn admission_race(webhook: bool) { + let state = crate::state::tests::test_state().await; + assert_eq!( + state.config.database_url, + std::env::var("BUZZ_TEST_DATABASE_URL").unwrap() + ); + let pool = sqlx::PgPool::connect(&state.config.database_url) + .await + .unwrap(); + let host = format!("admit-{}.example", Uuid::new_v4()); + let community = state + .db + .ensure_configured_community(&host) + .await + .unwrap() + .id; + let tenant = TenantContext::resolved(community, host.clone()); + let keys = Keys::generate(); + state + .db + .ensure_user(community, keys.public_key().as_bytes()) + .await + .unwrap(); + let channel = state + .db + .create_channel( + community, + "admit", + ChannelType::Stream, + ChannelVisibility::Private, + None, + keys.public_key().as_bytes(), + None, + ) + .await + .unwrap() + .id; + let id = Uuid::new_v4(); + let mut definition = serde_json::json!({"name":"admit", "enabled":true, + "trigger":{"on":"webhook"}, "steps":[{"id":"emit","action":"send_message","text":"old-action"}]}); + webhook_secret::inject_secret(&mut definition, "test-secret"); + let json = definition.to_string(); + state + .db + .upsert_workflow( + community, + id, + Some(channel), + keys.public_key().as_bytes(), + "admit", + &json, + &Sha256::digest(json.as_bytes()), + ) + .await + .unwrap(); + let sink = Arc::new(Sink::default()); + state.workflow_engine.set_action_sink(sink.clone()); + + for commit_update in [true, false] { + // Block the real authority SELECT after the caller selected its definition. + // Only then update: a caller that refreshes the record but executes the + // original definition will now incorrectly admit old-action. + let mut tx = pool.begin().await.unwrap(); + let pid: i32 = sqlx::query_scalar("SELECT pg_backend_pid()") + .fetch_one(&mut *tx) + .await + .unwrap(); + sqlx::query("LOCK TABLE channel_members IN ACCESS EXCLUSIVE MODE") + .execute(&mut *tx) + .await + .unwrap(); + let event = EventBuilder::new( + Kind::Custom(KIND_WORKFLOW_TRIGGER as u16), + Uuid::new_v4().to_string(), + ) + .tags([Tag::parse(["d", &id.to_string()]).unwrap()]) + .sign_with_keys(&keys) + .unwrap(); + let event_id = event.id; + let call_state = state.clone(); + let call_host = host.clone(); + let call_tenant = tenant.clone(); + let auth = IngestAuth::Nip42 { + pubkey: keys.public_key(), + scopes: vec![buzz_auth::Scope::MessagesWrite], + channel_ids: None, + conn_id: Uuid::new_v4(), + }; + let call = tokio::spawn(async move { + if webhook { + let mut headers = HeaderMap::new(); + headers.insert("host", call_host.parse().unwrap()); + headers.insert("x-webhook-secret", "test-secret".parse().unwrap()); + match crate::api::bridge::workflow_webhook( + State(call_state), + Path(id.to_string()), + Query(crate::api::bridge::WebhookQuery { secret: None }), + headers, + axum::body::Bytes::new(), + ) + .await + { + Ok((code, _)) => { + assert_eq!(code, StatusCode::ACCEPTED); + true + } + Err((code, _)) => { + assert_eq!(code, StatusCode::NOT_FOUND); + false + } + } + } else { + match handle_workflow_trigger(&call_tenant, &call_state, &event, &auth).await { + Ok(result) => { + assert!(result.accepted); + true + } + Err(IngestError::Rejected(reason)) => { + assert_eq!(reason, "forbidden: workflow changed or is no longer active"); + false + } + Err(e) => panic!("unexpected manual result: {e:?}"), + } + } + }); + tokio::time::timeout(Duration::from_secs(5), async { + loop { + assert!(!call.is_finished(), "caller did not reach authority gate"); + let waiting: bool = sqlx::query_scalar( + "SELECT EXISTS(SELECT 1 FROM pg_stat_activity \ + WHERE datname=current_database() AND query LIKE '%channel_members%' \ + AND $1=ANY(pg_blocking_pids(pid)))", + ) + .bind(pid) + .fetch_one(&pool) + .await + .unwrap(); + if waiting { + break; + } + tokio::time::sleep(Duration::from_millis(10)).await; + } + }) + .await + .expect("authority SELECT must block"); + definition["steps"][0]["text"] = serde_json::json!(if commit_update { + "new-action" + } else { + "rolled-back-action" + }); + let json = definition.to_string(); + buzz_db::workflow::upsert_workflow_in_transaction( + &mut tx, + community, + id, + Some(channel), + keys.public_key().as_bytes(), + "admit", + &json, + &Sha256::digest(json.as_bytes()), + ) + .await + .unwrap(); + if commit_update { + tx.commit().await.unwrap(); + } else { + tx.rollback().await.unwrap(); + } + let accepted = tokio::time::timeout(Duration::from_secs(5), call) + .await + .unwrap() + .unwrap(); + assert_eq!( + accepted, !commit_update, + "stale selection must not admit; rollback must admit" + ); + if commit_update { + assert!(state + .db + .list_workflow_runs(community, id, 100) + .await + .unwrap() + .is_empty()); + assert!(sink.0.lock().unwrap().is_empty()); + if !webhook { + let seen: bool = sqlx::query_scalar( + "SELECT EXISTS(SELECT 1 FROM events WHERE community_id=$1 AND id=$2)", + ) + .bind(community.as_uuid()) + .bind(event_id.as_bytes().as_slice()) + .fetch_one(&pool) + .await + .unwrap(); + assert!(!seen, "rejected manual command must roll back"); + } + } else { + tokio::time::timeout(Duration::from_secs(5), async { + loop { + let runs = state + .db + .list_workflow_runs(community, id, 100) + .await + .unwrap(); + assert_eq!(runs.len(), 1); + if runs[0].status == RunStatus::Completed { + break; + } + assert!(!matches!( + runs[0].status, + RunStatus::Failed | RunStatus::Cancelled + )); + tokio::time::sleep(Duration::from_millis(10)).await; + } + }) + .await + .expect("positive execution"); + assert_eq!(*sink.0.lock().unwrap(), ["new-action"]); + } + } +} + +#[tokio::test] +#[ignore = "requires PostgreSQL"] +async fn cluster_global_manual_admission_keeps_selected_revision_and_rejects_none() { + admission_race(false).await; +} +#[tokio::test] +#[ignore = "requires PostgreSQL"] +async fn cluster_global_webhook_admission_keeps_selected_revision_and_rejects_none() { + admission_race(true).await; +} diff --git a/crates/buzz-relay/src/handlers/ingest.rs b/crates/buzz-relay/src/handlers/ingest.rs index ee1d0312be9..64c8ed2b298 100644 --- a/crates/buzz-relay/src/handlers/ingest.rs +++ b/crates/buzz-relay/src/handlers/ingest.rs @@ -774,7 +774,10 @@ pub(crate) async fn check_channel_membership( } } -fn check_token_channel_access(auth: &IngestAuth, channel_id: Uuid) -> Result<(), String> { +pub(super) fn check_token_channel_access( + auth: &IngestAuth, + channel_id: Uuid, +) -> Result<(), String> { if let Some(allowed) = auth.channel_ids() { if !allowed.contains(&channel_id) { return Err("restricted: token does not have access to this channel".to_string()); @@ -2285,7 +2288,7 @@ async fn ingest_event_inner( // Command kinds are routed AFTER signature verification, timestamp check, // pubkey/auth match, and scope validation — never before. - if buzz_core::kind::is_command_kind(kind_u32) { + if buzz_core::kind::is_command_kind(kind_u32) && kind_u32 != KIND_WORKFLOW_DEF { return super::command_executor::handle_command(tenant, state, event, auth).await; } @@ -2398,6 +2401,19 @@ async fn ingest_event_inner( } } + // Workflow saves are content writes and must not bypass the durable ban/timeout gate. + if kind_u32 == KIND_WORKFLOW_DEF { + return super::command_executor::handle_command(tenant, state, event, auth).await; + } + + // Canonical workflow deletion owns storage and both projections atomically. + // Keep it after common signature, principal, scope and moderation checks, + // but before the generic address path (which has no resolved channel). + if let Some(coordinate) = super::workflow_lifecycle::deletion_coordinate(&event)? { + return super::workflow_lifecycle::delete(tenant, state, tracer, &event, &auth, coordinate) + .await; + } + let mut channel_id = if kind_u32 == KIND_REACTION { match derive_reaction_channel(tenant.community(), &state.db, &event).await { ReactionChannelResult::Channel(ch_id) => Some(ch_id), diff --git a/crates/buzz-relay/src/handlers/mod.rs b/crates/buzz-relay/src/handlers/mod.rs index 2f4aa00b595..f326aa4d9b7 100644 --- a/crates/buzz-relay/src/handlers/mod.rs +++ b/crates/buzz-relay/src/handlers/mod.rs @@ -40,6 +40,7 @@ pub mod report_resolution; pub mod req; /// NIP-29 and NIP-25 side-effect handlers. pub mod side_effects; +mod workflow_lifecycle; /// Extract an optional TTL (in seconds) from a Nostr event's `ttl` tag, /// applying the server-side override when configured. diff --git a/crates/buzz-relay/src/handlers/side_effects.rs b/crates/buzz-relay/src/handlers/side_effects.rs index 7282c913423..9aa7dfe6897 100644 --- a/crates/buzz-relay/src/handlers/side_effects.rs +++ b/crates/buzz-relay/src/handlers/side_effects.rs @@ -270,6 +270,12 @@ pub async fn validate_standard_deletion_event( .await? .ok_or_else(|| anyhow::anyhow!("target event not found"))?; + if u32::from(target_event.event.kind.as_u16()) == buzz_core::kind::KIND_WORKFLOW_DEF { + return Err(anyhow::anyhow!( + "workflow deletion requires a canonical a-tag coordinate" + )); + } + let target_author = effective_message_author(&target_event.event, &state.relay_keypair.public_key()); if target_author != actor_bytes @@ -595,6 +601,12 @@ pub async fn validate_admin_event( .map_err(|e| anyhow::anyhow!("db error looking up target: {e}"))? .ok_or_else(|| anyhow::anyhow!("target event not found"))?; + if u32::from(target_event.event.kind.as_u16()) == buzz_core::kind::KIND_WORKFLOW_DEF { + return Err(anyhow::anyhow!( + "workflow deletion requires an author-signed canonical a-tag coordinate" + )); + } + match target_event.channel_id { Some(target_ch) if target_ch != channel_id => { return Err(anyhow::anyhow!( @@ -2112,7 +2124,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 +2131,13 @@ 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}"); - } - } - } + // All workflow deletion belongs to the atomic pre-storage lifecycle. + // Do not retain a name/UUID runtime-only deletion escape hatch here. + return Err(anyhow::anyhow!( + "workflow deletion requires the atomic lifecycle path" + )); } - // 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 kinds keep their existing timestamp-ordered soft deletion. k if is_parameterized_replaceable(k) => { let pubkey_bytes = match hex::decode(pubkey_hex) { Ok(b) => b, diff --git a/crates/buzz-relay/src/handlers/workflow_lifecycle.rs b/crates/buzz-relay/src/handlers/workflow_lifecycle.rs new file mode 100644 index 00000000000..52bd7f0e6f8 --- /dev/null +++ b/crates/buzz-relay/src/handlers/workflow_lifecycle.rs @@ -0,0 +1,292 @@ +//! Canonical workflow deletion: author/channel validation and a single commit. + +use super::ingest::{IngestAuth, IngestError, IngestResult}; +use crate::state::AppState; +use buzz_core::{ + kind::{KIND_DELETION, KIND_WORKFLOW_DEF}, + TenantContext, +}; +use buzz_db::{ + workflow::{self, lifecycle}, + DbError, +}; +use nostr::Event; +use std::sync::Arc; +use uuid::Uuid; + +pub(super) struct Coordinate { + owner: [u8; 32], + id: Uuid, +} + +fn rejected(message: &str) -> IngestError { + IngestError::Rejected(message.into()) +} +fn db_error(error: DbError) -> IngestError { + IngestError::Internal(format!("error: workflow lifecycle: {error}")) +} + +/// Recognize every numeric workflow coordinate; reject aliases instead of falling through. +pub(super) fn deletion_coordinate(event: &Event) -> Result, IngestError> { + if u32::from(event.kind.as_u16()) != KIND_DELETION { + return Ok(None); + } + let targets: Vec<_> = event + .tags + .iter() + .filter(|t| matches!(t.kind().to_string().as_str(), "a" | "e")) + .collect(); + let Some(value) = targets.iter().find_map(|t| { + (t.kind().to_string() == "a") + .then(|| t.content()) + .flatten() + .filter(|v| { + v.split(':') + .next() + .and_then(|kind| kind.parse::().ok()) + == Some(KIND_WORKFLOW_DEF) + }) + }) else { + return Ok(None); + }; + let parts: Vec<_> = value.split(':').collect(); + if parts.len() != 3 || parts[0] != KIND_WORKFLOW_DEF.to_string() { + return Err(rejected("invalid: malformed workflow deletion coordinate")); + } + let id = Uuid::parse_str(parts[2]).map_err(|_| { + rejected("invalid: workflow deletion requires a canonical UUID, not a name") + })?; + if targets.len() != 1 || targets[0].as_slice().len() != 2 || parts[2] != id.to_string() { + return Err(rejected( + "invalid: workflow deletion requires exactly one canonical UUID coordinate", + )); + } + let owner = hex::decode(parts[1]) + .ok() + .and_then(|bytes| <[u8; 32]>::try_from(bytes).ok()) + .ok_or_else(|| rejected("invalid: malformed workflow owner"))?; + if parts[1] != hex::encode(owner) { + return Err(rejected("invalid: noncanonical workflow owner")); + } + Ok(Some(Coordinate { owner, id })) +} + +pub(super) async fn delete( + tenant: &TenantContext, + state: &Arc, + tracer: &Arc, + event: &Event, + auth: &IngestAuth, + coordinate: Coordinate, +) -> Result { + let Coordinate { owner, id } = coordinate; + if owner != auth.pubkey().to_bytes() || owner != event.pubkey.to_bytes() { + return Err(rejected( + "forbidden: only the workflow author can delete it", + )); + } + let community = tenant.community(); + let mut tx = state + .db + .begin_event_write_transaction() + .await + .map_err(db_error)?; + buzz_deletion::store(&state.db) + .guard_transaction(&mut tx, community) + .await + .map_err(|_| rejected("restricted: community writes are fenced"))?; + lifecycle::lock_coordinate(&mut tx, community, &owner, id) + .await + .map_err(db_error)?; + let head = lifecycle::head(&mut tx, community, &owner, id) + .await + .map_err(db_error)?; + let cutoff = lifecycle::deletion(&mut tx, community, &owner, id) + .await + .map_err(db_error)?; + let runtime = match workflow::get_workflow_in_transaction(&mut tx, community, id).await { + Ok(row) => Some(row), + Err(DbError::NotFound(_)) => None, + Err(error) => return Err(db_error(error)), + }; + if runtime + .as_ref() + .is_some_and(|row| row.owner_pubkey != owner) + { + return Err(rejected("forbidden: workflow belongs to a different owner")); + } + let channel = head + .as_ref() + .and_then(|h| h.channel_id) + .or_else(|| runtime.as_ref().and_then(|r| r.channel_id)) + .or_else(|| cutoff.as_ref().map(|d| d.channel_id)) + .ok_or_else(|| rejected("invalid: workflow not found"))?; + if head.as_ref().is_some_and(|h| h.channel_id != Some(channel)) + || runtime + .as_ref() + .is_some_and(|r| r.channel_id != Some(channel)) + || cutoff.as_ref().is_some_and(|d| d.channel_id != channel) + { + return Err(rejected("conflict: workflow lifecycle is unverified")); + } + let h_tags: Vec<_> = event + .tags + .iter() + .filter(|tag| tag.kind().to_string() == "h") + .collect(); + if h_tags.len() > 1 + || h_tags.first().is_some_and(|tag| { + tag.as_slice().len() != 2 || tag.content() != Some(channel.to_string().as_str()) + }) + { + return Err(rejected( + "forbidden: workflow deletion channel does not match", + )); + } + super::ingest::check_token_channel_access(auth, channel).map_err(IngestError::AuthFailed)?; + if lifecycle::channel_role(&mut tx, community, channel, &owner) + .await + .map_err(db_error)? + .is_none() + { + return Err(rejected( + "forbidden: workflow requires active channel membership", + )); + } + // Proof is forward-only; an old accepted event is not evidence that its side effects completed. + if lifecycle::event_seen(&mut tx, community, event) + .await + .map_err(db_error)? + { + if cutoff + .as_ref() + .is_some_and(|d| d.event_id == event.id.as_bytes().as_slice()) + { + emit_success(tracer, tenant, event, auth, channel, false); + return Ok(IngestResult { + event_id: event.id.to_hex(), + accepted: true, + message: "duplicate: workflow deletion previously committed".into(), + }); + } + return Err(rejected( + "conflict: previous workflow deletion is unverified; refresh required", + )); + } + if head.is_some() != runtime.is_some() { + return Err(rejected("conflict: workflow lifecycle is unverified")); + } + let timestamp = event.created_at.as_secs() as i64; + if head + .as_ref() + .is_some_and(|h| h.created_at.timestamp() > timestamp) + || cutoff + .as_ref() + .is_some_and(|d| d.deleted_through.timestamp() >= timestamp) + { + return Err(rejected( + "conflict: workflow deletion is stale; refresh required", + )); + } + let (stored, inserted) = + buzz_db::event::insert_event_in_transaction(&mut tx, community, event, Some(channel)) + .await + .map_err(db_error)?; + if !inserted { + return Err(rejected( + "conflict: previous workflow deletion is unverified; refresh required", + )); + } + lifecycle::delete_in_transaction(&mut tx, community, &owner, id, channel, event) + .await + .map_err(db_error)?; + tx.commit().await.map_err(|e| db_error(e.into()))?; + state + .workflow_engine + .invalidate_channel_workflows(community, channel); + super::event::dispatch_persistent_event( + tenant, + state, + &stored, + KIND_DELETION, + &event.pubkey.to_hex(), + None, + ) + .await; + emit_success(tracer, tenant, event, auth, channel, true); + Ok(IngestResult { + event_id: event.id.to_hex(), + accepted: true, + message: format!( + "response:{}", + serde_json::json!({"workflow_id":id,"deleted":true,"lifecycle_version":1}) + ), + }) +} + +fn emit_success( + tracer: &Arc, + tenant: &TenantContext, + event: &Event, + auth: &IngestAuth, + channel: Uuid, + inserted: bool, +) { + use crate::conformance::{ + channel_label, claimed_community_from_event, emit, msg_id_label, state_for_request, + TraceAction, + }; + let msg_id = msg_id_label(event.id.as_bytes()); + let channel = channel_label(channel); + let claimed_community = claimed_community_from_event(event); + let action = if inserted { + TraceAction::WriteInsert { + msg_id, + channel, + claimed_community, + } + } else { + TraceAction::WriteDuplicate { + msg_id, + channel, + claimed_community, + } + }; + emit(tracer, action, state_for_request(tenant, auth.pubkey())); +} + +#[cfg(test)] +mod postgres_tests; + +#[cfg(test)] +mod tests { + use super::*; + use buzz_core::kind::KIND_WORKFLOW_DEF; + use nostr::{EventBuilder, Keys, Kind, Tag}; + #[test] + fn routes_only_one_canonical_workflow_coordinate() { + let keys = Keys::generate(); + let id = Uuid::new_v4(); + let address = format!("{KIND_WORKFLOW_DEF}:{}:{id}", keys.public_key()); + let event = |tags| { + EventBuilder::new(Kind::EventDeletion, "") + .tags(tags) + .sign_with_keys(&keys) + .expect("fixture event") + }; + let a = Tag::parse(["a", &address]).expect("a tag"); + assert!(deletion_coordinate(&event(vec![a.clone()])) + .expect("valid") + .is_some()); + assert!(deletion_coordinate(&event(vec![a.clone(), a])).is_err()); + let alias = format!("{KIND_WORKFLOW_DEF}:{}:{}", keys.public_key(), id.simple()); + assert!( + deletion_coordinate(&event(vec![Tag::parse(["a", &alias]).expect("alias")])).is_err() + ); + assert!( + deletion_coordinate(&event(vec![Tag::parse(["e", &"a".repeat(64)]).expect("e")])) + .expect("generic") + .is_none() + ); + } +} diff --git a/crates/buzz-relay/src/handlers/workflow_lifecycle/postgres_tests.rs b/crates/buzz-relay/src/handlers/workflow_lifecycle/postgres_tests.rs new file mode 100644 index 00000000000..1bc67b25c71 --- /dev/null +++ b/crates/buzz-relay/src/handlers/workflow_lifecycle/postgres_tests.rs @@ -0,0 +1,769 @@ +//! Real ingest + PostgreSQL lifecycle tests. The nextest wrapper owns a database per test. +use super::super::ingest::ingest_event; +use super::*; +use buzz_auth::Scope; +use buzz_core::{ + channel::{ChannelType, ChannelVisibility}, + kind::KIND_WORKFLOW_DEF, +}; +use nostr::{EventBuilder, Keys, Kind, Tag, Timestamp}; + +struct Fixture { + state: Arc, + pool: sqlx::PgPool, + tenant: TenantContext, + keys: Keys, + channel: Uuid, + id: Uuid, + now: u64, + _cache: tempfile::TempDir, +} +impl Fixture { + async fn new() -> Self { + let url = + std::env::var("BUZZ_TEST_DATABASE_URL").expect("isolated PostgreSQL URL required"); + assert_eq!( + std::env::var("BUZZ_TEST_SCHEMA_MODE").as_deref(), + Ok("desired") + ); + let pool = sqlx::PgPool::connect(&url) + .await + .expect("connect isolated database"); + let db = buzz_db::Db::from_pool(pool.clone()); + let host = format!("workflow-{}.example", Uuid::new_v4()); + let community = db + .ensure_configured_community(&host) + .await + .expect("community") + .id; + let tenant = TenantContext::resolved(community, host); + let keys = Keys::generate(); + let channel = db + .create_channel( + community, + "workflow-test", + ChannelType::Stream, + ChannelVisibility::Private, + None, + keys.public_key().as_bytes(), + None, + ) + .await + .expect("channel") + .id; + let mut config = crate::config::Config::from_env().expect("config"); + config.database_url = url; + // Deliberately no live Redis, relay, S3, subscriber or workflow executor. + // The post-commit broadcast may fail; durable completion must not depend on it. + config.redis_url = "redis://127.0.0.1:1".into(); + config.require_relay_membership = false; + let cache = tempfile::tempdir().expect("cache dir"); + config.git_pack_cache_path = cache.path().to_path_buf(); + let redis = deadpool_redis::Config::from_url(&config.redis_url) + .create_pool(Some(deadpool_redis::Runtime::Tokio1)) + .expect("redis pool"); + let pubsub = Arc::new( + buzz_pubsub::PubSubManager::new(&config.redis_url, redis.clone()) + .await + .expect("pubsub"), + ); + let engine = Arc::new(buzz_workflow::WorkflowEngine::new( + db.clone(), + buzz_workflow::WorkflowConfig::default(), + )); + let auth = buzz_auth::AuthService::new(config.auth.clone()); + let search = buzz_search::SearchService::new(pool.clone()); + let media = buzz_media::MediaStorage::new(&config.media).expect("media config"); + let (state, _) = AppState::new( + config, + db, + redis, + None, + pubsub, + auth, + search, + engine, + Keys::generate(), + media, + ); + Self { + state: Arc::new(state), + pool, + tenant, + keys, + channel, + id: Uuid::new_v4(), + now: Timestamp::now().as_secs(), + _cache: cache, + } + } + fn auth(&self) -> IngestAuth { + IngestAuth::Nip42 { + pubkey: self.keys.public_key(), + scopes: vec![Scope::MessagesWrite], + channel_ids: None, + conn_id: Uuid::new_v4(), + } + } + fn save(&self, time: u64, name: &str) -> Event { + self.sign(Kind::Custom(KIND_WORKFLOW_DEF as u16), time, + &format!("name: {name}\nenabled: false\ntrigger:\n on: message_posted\nsteps:\n - id: wait\n action: delay\n duration: 1s\n"), + vec![vec!["d".into(), self.id.to_string()], vec!["h".into(), self.channel.to_string()]]) + } + fn delete(&self, time: u64) -> Event { + self.sign( + Kind::EventDeletion, + time, + "", + vec![vec![ + "a".into(), + format!("30620:{}:{}", self.keys.public_key(), self.id), + ]], + ) + } + fn sign(&self, kind: Kind, time: u64, content: &str, tags: Vec>) -> Event { + EventBuilder::new(kind, content) + .tags(tags.into_iter().map(|t| Tag::parse(t).expect("tag"))) + .custom_created_at(Timestamp::from(time)) + .sign_with_keys(&self.keys) + .expect("signed fixture") + } + async fn send(&self, event: &Event) -> Result { + ingest_event(&self.state, &self.tenant, event.clone(), self.auth()).await + } + async fn live(&self) -> (Option>, Option, Option>) { + let mut tx = self + .state + .db + .begin_event_write_transaction() + .await + .expect("tx"); + let owner = self.keys.public_key().to_bytes(); + let head = lifecycle::head(&mut tx, self.tenant.community(), &owner, self.id) + .await + .expect("head"); + let cutoff = lifecycle::deletion(&mut tx, self.tenant.community(), &owner, self.id) + .await + .expect("cutoff"); + let runtime = + match workflow::get_workflow_in_transaction(&mut tx, self.tenant.community(), self.id) + .await + { + Ok(row) => Some(row.name), + Err(DbError::NotFound(_)) => None, + Err(e) => panic!("runtime: {e}"), + }; + (head.map(|h| h.id), runtime, cutoff.map(|d| d.event_id)) + } + async fn seen(&self, event: &Event) -> bool { + let mut tx = self + .state + .db + .begin_event_write_transaction() + .await + .expect("tx"); + lifecycle::event_seen(&mut tx, self.tenant.community(), event) + .await + .expect("seen") + } +} +fn reject(result: Result, prefix: &str) { + match result { + Err(IngestError::Rejected(message) | IngestError::AuthFailed(message)) => { + assert!(message.starts_with(prefix), "{message}") + } + Err(other) => panic!("expected {prefix}, got {other:?}"), + Ok(result) => panic!("expected {prefix}, got success {}", result.message), + } +} + +#[tokio::test] +#[ignore = "requires PostgreSQL"] +async fn ingest_save_projects_enabled_for_creation_updates_and_default() { + let f = Fixture::new().await; + let mut previous = None; + for (index, configured) in [Some(false), Some(true), Some(false), None, Some(false)] + .into_iter() + .enumerate() + { + let enabled = configured.unwrap_or(true); + let setting = configured + .map(|value| format!("enabled: {value}\n")) + .unwrap_or_default(); + let yaml = format!("name: enabled projection\n{setting}trigger:\n on: reaction_added\nsteps:\n - id: wait\n action: delay\n duration: 1s\n"); + let mut tags = vec![ + vec!["d".into(), f.id.to_string()], + vec!["h".into(), f.channel.to_string()], + ]; + if let Some(revision) = previous { + tags.push(vec!["expected-revision".into(), revision]); + } + let event = f.sign( + Kind::Custom(KIND_WORKFLOW_DEF as u16), + f.now + index as u64, + &yaml, + tags, + ); + assert!( + f.send(&event) + .await + .expect("save configured state") + .accepted + ); + let row = f + .state + .db + .get_workflow(f.tenant.community(), f.id) + .await + .expect("runtime"); + assert_eq!( + row.definition["enabled"], enabled, + "canonical definition step {index}" + ); + assert_eq!( + row.enabled, enabled, + "runtime enabled projection step {index}" + ); + assert_eq!(f.live().await.0, Some(event.id.to_bytes().to_vec())); + let eligible = f + .state + .db + .list_enabled_channel_workflows(f.tenant.community(), f.channel) + .await + .expect("trigger eligibility"); + assert_eq!( + eligible.iter().any(|row| row.id == f.id), + enabled, + "automatic eligibility step {index}" + ); + if !enabled { + let trigger = f.sign( + Kind::Custom(buzz_core::kind::KIND_WORKFLOW_TRIGGER as u16), + f.now + 20 + index as u64, + "", + vec![ + vec!["d".into(), f.id.to_string()], + vec!["h".into(), f.channel.to_string()], + ], + ); + reject(f.send(&trigger).await, "forbidden: workflow is disabled"); + assert!(!f.seen(&trigger).await); + } + previous = Some(event.id.to_hex()); + } +} + +#[tokio::test] +#[ignore = "requires PostgreSQL"] +async fn ingest_save_delete_replay_and_both_timestamp_orders() { + let f = Fixture::new().await; + let create = f.save(f.now, "first"); + assert!(f.send(&create).await.expect("create").accepted); + assert_eq!( + f.live().await, + ( + Some(create.id.to_bytes().to_vec()), + Some("first".into()), + None + ) + ); + let stale = f.delete(f.now - 1); + reject(f.send(&stale).await, "conflict:"); + assert!(!f.seen(&stale).await); + // NIP-09 includes the same second; the arrival-order reversal below must agree. + let delete = f.delete(f.now); + let result = f.send(&delete).await.expect("delete"); + assert!(result.message.contains("\"deleted\":true")); + assert_eq!( + f.live().await, + (None, None, Some(delete.id.to_bytes().to_vec())) + ); + assert!(f + .send(&delete) + .await + .expect("verified replay") + .message + .starts_with("duplicate:")); + assert!(f + .send(&create) + .await + .expect("exact save replay") + .message + .starts_with("duplicate:")); + for time in [f.now - 1, f.now] { + let delayed = f.save(time, "unseen delayed save"); + reject(f.send(&delayed).await, "conflict:"); + assert!(!f.seen(&delayed).await); + } + let newer = f.save(f.now + 1, "newer"); + assert!( + f.send(&newer) + .await + .expect("intentional recreation") + .accepted + ); + assert!(f + .send(&delete) + .await + .expect("old delete replay") + .message + .starts_with("duplicate:")); + assert_eq!( + f.live().await, + ( + Some(newer.id.to_bytes().to_vec()), + Some("newer".into()), + Some(delete.id.to_bytes().to_vec()) + ) + ); +} + +#[tokio::test] +#[ignore = "requires PostgreSQL"] +async fn ingest_rolls_back_both_projections_and_proof_on_commit_failure() { + let f = Fixture::new().await; + let create = f.save(f.now, "first"); + f.send(&create).await.expect("create"); + let before = f.live().await; + // A deferred runtime trigger fails at COMMIT, after the real save upsert. + // The former pool-based upsert would fail here before the event commit; the + // delete control additionally catches an event/proof committed before runtime work. + sqlx::raw_sql("CREATE FUNCTION fail_workflow_commit() RETURNS trigger LANGUAGE plpgsql AS $$ BEGIN RAISE EXCEPTION 'fixture commit failure'; END $$; CREATE CONSTRAINT TRIGGER fail_workflow_commit AFTER INSERT OR UPDATE OR DELETE ON workflows DEFERRABLE INITIALLY DEFERRED FOR EACH ROW EXECUTE FUNCTION fail_workflow_commit();") + .execute(&f.pool).await.expect("fault trigger"); + let update = f.save(f.now + 1, "failed update"); + assert!(matches!( + f.send(&update).await, + Err(IngestError::Internal(_)) + )); + assert_eq!(f.live().await, before); + assert!(!f.seen(&update).await); + let delete = f.delete(f.now + 2); + assert!(matches!( + f.send(&delete).await, + Err(IngestError::Internal(_)) + )); + assert_eq!(f.live().await, before); + assert!(!f.seen(&delete).await); + sqlx::raw_sql("DROP TRIGGER fail_workflow_commit ON workflows") + .execute(&f.pool) + .await + .expect("clear fault"); + f.send(&update) + .await + .expect("same event retry after rollback"); + f.send(&delete) + .await + .expect("same delete retry after rollback"); + assert_eq!( + f.live().await, + (None, None, Some(delete.id.to_bytes().to_vec())) + ); +} + +#[tokio::test] +#[ignore = "requires PostgreSQL"] +async fn ingest_denies_wrong_channel_token_owner_revocation_and_legacy_completion() { + let f = Fixture::new().await; + let create = f.save(f.now, "first"); + f.send(&create).await.expect("create"); + let before = f.live().await; + let delete = f.delete(f.now + 1); + let mut scoped = f.auth(); + if let IngestAuth::Nip42 { channel_ids, .. } = &mut scoped { + *channel_ids = Some(vec![Uuid::new_v4()]); + } + reject( + ingest_event(&f.state, &f.tenant, delete.clone(), scoped.clone()).await, + "restricted:", + ); + reject( + ingest_event(&f.state, &f.tenant, f.save(f.now + 1, "restricted"), scoped).await, + "restricted:", + ); + let wrong_h = f.sign( + Kind::EventDeletion, + f.now + 1, + "", + vec![ + vec![ + "a".into(), + format!("30620:{}:{}", f.keys.public_key(), f.id), + ], + vec!["h".into(), Uuid::new_v4().to_string()], + ], + ); + reject(f.send(&wrong_h).await, "forbidden:"); + let foreign = Keys::generate(); + let wrong_owner = f.sign( + Kind::EventDeletion, + f.now + 1, + "", + vec![vec![ + "a".into(), + format!("30620:{}:{}", foreign.public_key(), f.id), + ]], + ); + reject(f.send(&wrong_owner).await, "forbidden:"); + // An old accepted generic tombstone is not a forward completion proof. + let mut tx = f + .state + .db + .begin_event_write_transaction() + .await + .expect("tx"); + buzz_db::event::insert_event_in_transaction( + &mut tx, + f.tenant.community(), + &delete, + Some(f.channel), + ) + .await + .expect("legacy event"); + tx.commit().await.expect("legacy commit"); + reject(f.send(&delete).await, "conflict:"); + assert_eq!(f.live().await, before); + sqlx::query( + "UPDATE channel_members SET removed_at=NOW() WHERE community_id=$1 AND channel_id=$2", + ) + .bind(f.tenant.community().as_uuid()) + .bind(f.channel) + .execute(&f.pool) + .await + .expect("revoke"); + reject(f.send(&f.delete(f.now + 2)).await, "forbidden:"); + reject(f.send(&f.save(f.now + 2, "revoked")).await, "forbidden:"); + assert_eq!(f.live().await, before); +} + +#[tokio::test] +#[ignore = "requires PostgreSQL"] +async fn ingest_save_event_commit_failure_cannot_leave_pool_runtime_upsert() { + let f = Fixture::new().await; + let create = f.save(f.now, "first"); + f.send(&create).await.expect("create"); + let before = f.live().await; + // Events are partitioned: install the deferred failure on the real partition. + sqlx::raw_sql("CREATE FUNCTION fail_event_commit() RETURNS trigger LANGUAGE plpgsql AS $$ BEGIN RAISE EXCEPTION 'fixture event commit failure'; END $$; DO $$ DECLARE partition regclass; BEGIN FOR partition IN SELECT inhrelid::regclass FROM pg_inherits WHERE inhparent='events'::regclass LOOP EXECUTE format('CREATE CONSTRAINT TRIGGER fail_event_commit AFTER INSERT ON %s DEFERRABLE INITIALLY DEFERRED FOR EACH ROW EXECUTE FUNCTION fail_event_commit()', partition); END LOOP; END $$;") + .execute(&f.pool).await.expect("event fault trigger"); + let update = f.save(f.now + 1, "must roll back runtime too"); + assert!(matches!( + f.send(&update).await, + Err(IngestError::Internal(_)) + )); + assert_eq!(f.live().await, before); + assert!(!f.seen(&update).await); +} + +#[tokio::test] +#[ignore = "requires PostgreSQL"] +async fn ingest_concurrent_save_delete_share_coordinate_lock_and_keep_newer_head() { + let f = Fixture::new().await; + f.send(&f.save(f.now, "first")).await.expect("create"); + let mut blocker = f + .state + .db + .begin_event_write_transaction() + .await + .expect("blocker"); + lifecycle::lock_coordinate( + &mut blocker, + f.tenant.community(), + f.keys.public_key().as_bytes(), + f.id, + ) + .await + .expect("coordinate lock"); + let save = f.save(f.now + 2, "newer wins"); + let delete = f.delete(f.now + 1); + let (save_result, delete_result) = tokio::join!(async { f.send(&save).await }, async { + let delete_future = f.send(&delete); + let release = async { + tokio::time::timeout(std::time::Duration::from_secs(5), async { + loop { + let waiting: i64 = sqlx::query_scalar("SELECT count(*) FROM pg_locks WHERE locktype='advisory' AND NOT granted AND database=(SELECT oid FROM pg_database WHERE datname=current_database())") + .fetch_one(&f.pool).await.expect("lock waiters"); + if waiting >= 2 { break; } + tokio::time::sleep(std::time::Duration::from_millis(10)).await; + } + }).await.expect("both production paths must wait on the coordinate lock"); + blocker.commit().await.expect("release lock"); + }; + let (result, ()) = tokio::join!(delete_future, release); + result + }); + assert!(save_result.expect("save").accepted); + match delete_result { + Ok(r) => assert!(r.accepted), + Err(IngestError::Rejected(m)) => assert!(m.starts_with("conflict:")), + other => panic!("unexpected delete: {}", other.is_ok()), + } + let (head, runtime, _) = f.live().await; + assert_eq!(head, Some(save.id.to_bytes().to_vec())); + assert_eq!(runtime, Some("newer wins".into())); +} + +#[tokio::test] +#[ignore = "requires PostgreSQL"] +async fn ingest_rejects_legacy_split_missing_archived_and_cross_community_targets() { + let f = Fixture::new().await; + reject(f.send(&f.delete(f.now)).await, "invalid:"); + let create = f.save(f.now, "first"); + f.send(&create).await.expect("create"); + let other_host = format!("other-{}.example", Uuid::new_v4()); + let other_community = f + .state + .db + .ensure_configured_community(&other_host) + .await + .expect("other") + .id; + let other = TenantContext::resolved(other_community, other_host); + reject( + ingest_event(&f.state, &other, f.delete(f.now + 1), f.auth()).await, + "invalid:", + ); + reject( + ingest_event( + &f.state, + &other, + f.save(f.now + 1, "wrong tenant"), + f.auth(), + ) + .await, + "forbidden:", + ); + let before = f.live().await; + sqlx::query("UPDATE channels SET archived_at=NOW() WHERE community_id=$1 AND id=$2") + .bind(f.tenant.community().as_uuid()) + .bind(f.channel) + .execute(&f.pool) + .await + .expect("archive"); + reject(f.send(&f.delete(f.now + 1)).await, "forbidden:"); + reject(f.send(&f.save(f.now + 1, "archived")).await, "forbidden:"); + assert_eq!(f.live().await, before); + sqlx::query("UPDATE channels SET archived_at=NULL WHERE community_id=$1 AND id=$2") + .bind(f.tenant.community().as_uuid()) + .bind(f.channel) + .execute(&f.pool) + .await + .expect("unarchive"); + sqlx::query("DELETE FROM workflows WHERE community_id=$1 AND id=$2") + .bind(f.tenant.community().as_uuid()) + .bind(f.id) + .execute(&f.pool) + .await + .expect("legacy split"); + reject(f.send(&f.delete(f.now + 1)).await, "conflict:"); + assert_eq!( + f.live().await, + (Some(create.id.to_bytes().to_vec()), None, None) + ); +} + +#[tokio::test] +#[ignore = "requires PostgreSQL"] +async fn ingest_ban_timeout_and_schema_fence_are_enforced() { + let f = Fixture::new().await; + f.send(&f.save(f.now, "first")).await.expect("create"); + for (banned, until) in [ + (true, None), + ( + false, + Some(chrono::Utc::now() + chrono::Duration::minutes(5)), + ), + ] { + sqlx::query("INSERT INTO community_bans (community_id,pubkey,banned,muted_until,actor_pubkey) VALUES ($1,$2,$3,$4,$2) ON CONFLICT (community_id,pubkey) DO UPDATE SET banned=$3,muted_until=$4") + .bind(f.tenant.community().as_uuid()).bind(f.keys.public_key().as_bytes().as_slice()).bind(banned).bind(until).execute(&f.pool).await.expect("restriction"); + let prefix = if banned { "blocked:" } else { "restricted:" }; + reject(f.send(&f.save(f.now + 1, "blocked")).await, prefix); + reject(f.send(&f.delete(f.now + 1)).await, prefix); + } + let attached: bool = sqlx::query_scalar("SELECT EXISTS(SELECT 1 FROM pg_trigger WHERE tgrelid='workflow_deletions'::regclass AND NOT tgisinternal AND tgfoid='enforce_community_write_fence()'::regprocedure)") + .fetch_one(&f.pool).await.expect("live fence catalog"); + assert!( + attached, + "desired-state deletion proof table must have a real write fence" + ); +} + +#[tokio::test] +#[ignore = "requires PostgreSQL"] +async fn ingest_concurrent_webhook_saves_preserve_one_secret_and_cas_revision() { + let f = Fixture::new().await; + f.send(&f.save(f.now, "first")).await.expect("create"); + let webhook = |time, name| { + let save = f.save(time, name); + f.sign( + save.kind, + time, + &save.content.replace("on: message_posted", "on: webhook"), + save.tags.iter().map(|t| t.as_slice().to_vec()).collect(), + ) + }; + let first = webhook(f.now + 1, "webhook first"); + let second = webhook(f.now + 2, "webhook second"); + let (a, b) = tokio::join!(f.send(&first), f.send(&second)); + let receipts: Vec = [a.expect("first outcome"), b.expect("second outcome")] + .iter() + .filter_map(|r| { + r.message + .strip_prefix("response:") + .map(|s| serde_json::from_str(s).expect("receipt")) + }) + .collect(); + let secrets: Vec<_> = receipts + .iter() + .filter_map(|v| v.get("webhook_secret").and_then(|v| v.as_str())) + .collect(); + assert_eq!( + secrets.len(), + 1, + "only the first committed webhook transition gets a secret" + ); + let row = f + .state + .db + .get_workflow(f.tenant.community(), f.id) + .await + .expect("runtime"); + assert_eq!( + crate::webhook_secret::extract_secret(&row.definition).as_deref(), + Some(secrets[0]) + ); + assert_eq!(row.name, "webhook second"); + let stale = f.sign( + first.kind, + f.now + 3, + &first.content, + vec![ + vec!["h".into(), f.channel.to_string()], + vec!["d".into(), f.id.to_string()], + vec!["expected-revision".into(), first.id.to_hex()], + ], + ); + reject(f.send(&stale).await, "conflict:"); + assert_eq!(f.live().await.0, Some(second.id.to_bytes().to_vec())); +} + +#[tokio::test] +#[ignore = "requires PostgreSQL"] +async fn ingest_rejects_alternate_workflow_deletion_entrances_before_storage() { + let f = Fixture::new().await; + let create = f.save(f.now, "named-helper"); + f.send(&create).await.expect("create"); + let before = f.live().await; + let owner = f.keys.public_key(); + for address in [ + format!("30620:{owner}:named-helper"), + format!("030620:{owner}:{}", f.id), + format!("+30620:{owner}:{}", f.id), + format!("30620:{owner}:{}", f.id.simple()), + ] { + let delete = f.sign( + Kind::EventDeletion, + f.now + 1, + "", + vec![vec!["a".into(), address]], + ); + assert!( + f.send(&delete).await.is_err(), + "noncanonical workflow delete must reject" + ); + assert!( + !f.seen(&delete).await, + "rejection must precede durable acceptance" + ); + assert_eq!(f.live().await, before); + } + for kind in [Kind::EventDeletion, Kind::Custom(9005)] { + let delete = f.sign( + kind, + f.now + 1, + "", + vec![ + vec!["e".into(), create.id.to_hex()], + vec!["h".into(), f.channel.to_string()], + ], + ); + assert!( + f.send(&delete).await.is_err(), + "definition-only deletion must reject" + ); + assert!(!f.seen(&delete).await); + assert_eq!(f.live().await, before); + } + // Generic message deletion still works through each original path. + for (index, kind) in [Kind::EventDeletion, Kind::Custom(9005)] + .into_iter() + .enumerate() + { + let message = f.sign( + Kind::Custom(9), + f.now + index as u64, + "ordinary message", + vec![vec!["h".into(), f.channel.to_string()]], + ); + f.send(&message).await.expect("message create"); + let delete = f.sign( + kind, + f.now + 3, + "", + vec![ + vec!["e".into(), message.id.to_hex()], + vec!["h".into(), f.channel.to_string()], + ], + ); + assert!(f.send(&delete).await.expect("ordinary delete").accepted); + let live: bool = sqlx::query_scalar("SELECT EXISTS(SELECT 1 FROM events WHERE community_id=$1 AND id=$2 AND deleted_at IS NULL)") + .bind(f.tenant.community().as_uuid()).bind(message.id.as_bytes().as_slice()) + .fetch_one(&f.pool).await.expect("live message"); + assert!(!live, "ordinary message was deleted"); + assert_eq!(f.live().await, before, "workflow remains untouched"); + } +} + +#[tokio::test] +#[ignore = "requires isolated PostgreSQL"] +async fn nip11_workflow_contract_requires_bound_host_and_stable_identity() { + let mut f = Fixture::new().await; + // No stable key: no forward compatibility promise, even on a mapped host. + let state = Arc::get_mut(&mut f.state).expect("sole state owner"); + Arc::make_mut(&mut state.config).advertise_workflow_lifecycle = true; + Arc::make_mut(&mut state.config).relay_private_key = None; + let info = crate::nip11::nip11_document(&f.state, f.tenant.host()).await; + assert!(info.workflows.is_none()); + let state = Arc::get_mut(&mut f.state).expect("sole state owner"); + Arc::make_mut(&mut state.config).relay_private_key = + Some(state.relay_keypair.secret_key().to_secret_hex()); + let info = crate::nip11::nip11_document(&f.state, f.tenant.host()).await; + let descriptor = info.workflows.expect("mapped stable relay advertises"); + assert_eq!(descriptor.lifecycle, 1); + assert_eq!(descriptor.host, f.tenant.host()); + assert_eq!( + info.relay_self, + Some(f.state.relay_keypair.public_key().to_hex()) + ); + assert!(info + .supported_extensions + .expect("extensions") + .contains(&"buzz-workflows".into())); + for host in ["", "unmapped-workflow.invalid"] { + let info = crate::nip11::nip11_document(&f.state, host).await; + assert!(info.workflows.is_none(), "unmapped host must not advertise"); + assert!(!info + .supported_extensions + .expect("extensions") + .contains(&"buzz-workflows".into())); + } + f.pool.close().await; + let info = crate::nip11::nip11_document(&f.state, f.tenant.host()).await; + assert!( + info.workflows.is_none(), + "failed binding must not advertise" + ); +} + +mod report_delete_postgres_tests; + +mod stale_execution_postgres_tests; diff --git a/crates/buzz-relay/src/handlers/workflow_lifecycle/postgres_tests/report_delete_postgres_tests.rs b/crates/buzz-relay/src/handlers/workflow_lifecycle/postgres_tests/report_delete_postgres_tests.rs new file mode 100644 index 00000000000..7a7aff7f4d0 --- /dev/null +++ b/crates/buzz-relay/src/handlers/workflow_lifecycle/postgres_tests/report_delete_postgres_tests.rs @@ -0,0 +1,240 @@ +//! Report-resolution and recovery must not split a freshly saved workflow. +use super::*; +use crate::handlers::{admin_action_worker, report_resolution}; + +async fn report( + f: &Fixture, + target: &Event, + time: u64, +) -> buzz_db::admin_moderation::AdminReportDetail { + let event = EventBuilder::new(Kind::Custom(1984), "workflow report") + .tags([ + Tag::parse(vec!["e".into(), target.id.to_hex(), "spam".into()]).expect("e tag"), + Tag::parse(vec!["p".into(), target.pubkey.to_hex()]).expect("p tag"), + ]) + .allow_self_tagging() + .custom_created_at(Timestamp::from(time)) + .sign_with_keys(&f.keys) + .expect("signed report"); + assert!(f.send(&event).await.expect("ingest report").accepted); + let id: Uuid = sqlx::query_scalar( + "SELECT id FROM moderation_reports WHERE community_id = $1 AND report_event_id = $2", + ) + .bind(f.tenant.community().as_uuid()) + .bind(event.id.as_bytes().as_slice()) + .fetch_one(&f.pool) + .await + .expect("stored report"); + f.state + .db + .admin_get_report(id) + .await + .expect("report lookup") + .expect("report exists") +} + +async fn resolve( + f: &Fixture, + report: &buzz_db::admin_moderation::AdminReportDetail, + recover: bool, +) -> Uuid { + let actor = Keys::generate().public_key().to_bytes(); + let request_id = Uuid::new_v4(); + if !recover { + // This is the production orchestration invoked by authorized HTTP report resolution. + // It is intentionally not a test of HTTP authentication. + let result = report_resolution::resolve_report_with_enforcement( + &f.state, + &f.tenant, + report, + "delete", + None, + None, + request_id, + &actor, + "operator", + "relay_operator", + ) + .await; + return match result { + Ok(done) => done.action_id, + Err(report_resolution::ResolutionError::EnforcementFailed { action_id, .. }) => { + action_id + } + Err(other) => panic!("unexpected resolution failure: {other:?}"), + }; + } + // A crash after the durable claim but before mutation must take the same guarded path. + let (owner, target) = report_resolution::derive_enforcement_target(report).expect("target"); + let claim = f + .state + .db + .claim_report_for_enforcement( + f.tenant.community(), + report.report.id, + request_id, + &actor, + "operator", + "delete", + None, + None, + "resolve:delete", + "relay_operator", + owner.as_deref(), + target.as_deref(), + report.report.channel_id, + ) + .await + .expect("claim before crash"); + let buzz_db::relay_admin_actions::ClaimResult::Claimed(action) = claim else { + panic!("fresh report must be claimed"); + }; + let mut batch = f + .state + .db + .claim_stranded_admin_action_batch( + "workflow-report-regression", + chrono::Utc::now() + chrono::Duration::seconds(120), + 8, + ) + .await + .expect("recovery batch"); + assert_eq!( + batch.len(), + 1, + "isolated database contains only this stranded action" + ); + let stranded = batch.remove(0); + assert_eq!(stranded.record.id, action.id); + admin_action_worker::recover_one(&f.state, stranded).await; + action.id +} + +async fn assert_workflow_rejected(f: &Fixture, action_id: Uuid) { + let action = f + .state + .db + .get_admin_action(action_id) + .await + .expect("action") + .expect("exists"); + assert_eq!( + action.state, "failed", + "report delete must fail rather than splitting lifecycle" + ); + assert_eq!(action.step_marker, None, "no successful mutation marker"); + assert!(action + .error_message + .as_deref() + .expect("explicit failure") + .contains("workflow definitions require canonical author-signed deletion")); + let outbox: i64 = + sqlx::query_scalar("SELECT COUNT(*) FROM relay_admin_outbox WHERE action_id = $1") + .bind(action_id) + .fetch_one(&f.pool) + .await + .expect("outbox count"); + assert_eq!( + outbox, 0, + "do not announce an enforcement that never happened" + ); +} + +async fn workflow_report_delete(recover: bool) { + let f = Fixture::new().await; + let disabled = f.save(f.now, "report target"); + let saved = f.sign( + Kind::Custom(KIND_WORKFLOW_DEF as u16), + f.now, + &disabled.content.replace("enabled: false", "enabled: true"), + vec![ + vec!["d".into(), f.id.to_string()], + vec!["h".into(), f.channel.to_string()], + ], + ); + assert!(f.send(&saved).await.expect("atomic save").accepted); + let before = f.live().await; + let reported = report(&f, &saved, f.now + 1).await; + let action_id = resolve(&f, &reported, recover).await; + assert_workflow_rejected(&f, action_id).await; + assert_eq!( + f.live().await, + before, + "definition, runtime and cutoff must stay unchanged" + ); + let enabled = f + .state + .db + .list_enabled_channel_workflows(f.tenant.community(), f.channel) + .await + .expect("runtime selection"); + assert!(enabled.iter().any(|row| row.id == f.id)); + // File a second report while the target is live, then resolve it after the + // owner's canonical deletion. Report ingest correctly rejects hidden targets. + let pending = report(&f, &saved, f.now + 2).await; + let deletion = f.delete(f.now + 3); + assert!( + f.send(&deletion) + .await + .expect("canonical deletion remains usable") + .accepted + ); + assert_eq!( + f.live().await, + (None, None, Some(deletion.id.to_bytes().to_vec())) + ); + // A pending report against a tombstoned definition must not acquire a success marker. + let repeated_id = resolve(&f, &pending, recover).await; + assert_workflow_rejected(&f, repeated_id).await; + assert_eq!( + f.live().await, + (None, None, Some(deletion.id.to_bytes().to_vec())) + ); +} + +#[tokio::test] +#[ignore = "requires PostgreSQL"] +async fn report_resolution_rejects_workflow_delete_and_preserves_owner_recovery() { + workflow_report_delete(false).await; +} + +#[tokio::test] +#[ignore = "requires PostgreSQL"] +async fn report_recovery_worker_rejects_workflow_delete_and_preserves_owner_recovery() { + workflow_report_delete(true).await; +} + +#[tokio::test] +#[ignore = "requires PostgreSQL"] +async fn report_deletion_still_enforces_ordinary_events_through_both_drivers() { + let f = Fixture::new().await; + for (i, recover) in [false, true].into_iter().enumerate() { + let event = f.sign( + Kind::Custom(9), + f.now + i as u64, + "ordinary content", + vec![vec!["h".into(), f.channel.to_string()]], + ); + assert!(f.send(&event).await.expect("message").accepted); + let reported = report(&f, &event, f.now + 10 + i as u64).await; + let action_id = resolve(&f, &reported, recover).await; + let action = f + .state + .db + .get_admin_action(action_id) + .await + .expect("action") + .expect("exists"); + assert_eq!(action.state, "succeeded"); + assert_eq!(action.step_marker.as_deref(), Some("artifacts_done")); + let deleted: bool = sqlx::query_scalar( + "SELECT deleted_at IS NOT NULL FROM events WHERE community_id = $1 AND id = $2", + ) + .bind(f.tenant.community().as_uuid()) + .bind(event.id.as_bytes().as_slice()) + .fetch_one(&f.pool) + .await + .expect("stored target"); + assert!(deleted, "ordinary event moderation remains effective"); + } +} diff --git a/crates/buzz-relay/src/handlers/workflow_lifecycle/postgres_tests/stale_execution_postgres_tests.rs b/crates/buzz-relay/src/handlers/workflow_lifecycle/postgres_tests/stale_execution_postgres_tests.rs new file mode 100644 index 00000000000..082b15c4fe2 --- /dev/null +++ b/crates/buzz-relay/src/handlers/workflow_lifecycle/postgres_tests/stale_execution_postgres_tests.rs @@ -0,0 +1,310 @@ +//! Two independent engine caches over real signed lifecycle ingest and PostgreSQL. +use super::*; +use buzz_db::workflow::RunStatus; +use buzz_workflow::{ActionSink, ActionSinkError, WorkflowConfig, WorkflowEngine}; +use std::{ + future::Future, + pin::Pin, + sync::Mutex, + time::{Duration, Instant}, +}; + +#[derive(Default)] +struct RecordingSink(Mutex>); +impl ActionSink for RecordingSink { + fn send_message( + &self, + _community_id: buzz_core::CommunityId, + _channel_id: &str, + text: &str, + _authored_text: &str, + _author_pubkey: &str, + _reply_to: Option<&str>, + ) -> Pin> + Send + '_>> { + self.0.lock().expect("sink lock").push(text.to_owned()); + Box::pin(async { Ok("ab".repeat(32)) }) + } +} + +fn definition(f: &Fixture, time: u64, enabled: bool, text: &str) -> Event { + f.sign( + Kind::Custom(KIND_WORKFLOW_DEF as u16), time, + &format!("name: cache fence\nenabled: {enabled}\ntrigger:\n on: reaction_added\nsteps:\n - id: emit\n action: send_message\n text: {text}\n"), + vec![vec!["d".into(), f.id.to_string()], vec!["h".into(), f.channel.to_string()]], + ) +} + +fn reaction(f: &Fixture) -> buzz_core::StoredEvent { + buzz_core::StoredEvent::new( + f.sign( + Kind::Reaction, + f.now, + "+", + vec![vec!["h".into(), f.channel.to_string()]], + ), + Some(f.channel), + ) +} + +async fn runs(f: &Fixture) -> Vec { + f.state + .db + .list_workflow_runs(f.tenant.community(), f.id, 100) + .await + .expect("runs") +} + +async fn completed(f: &Fixture, count: usize) { + tokio::time::timeout(Duration::from_secs(5), async { + loop { + let rows = runs(f).await; + assert_eq!(rows.len(), count, "unexpected run count"); + assert!( + rows.iter() + .all(|r| !matches!(r.status, RunStatus::Failed | RunStatus::Cancelled)), + "execution failed: {rows:?}" + ); + if rows.iter().all(|r| r.status == RunStatus::Completed) { + break; + } + tokio::task::yield_now().await; + } + }) + .await + .expect("executions settle"); +} + +#[derive(Clone, Copy, Debug)] +enum Transition { + Update, + Disable, + Recreate, + RecreateIdentical, +} + +async fn stale_engine_cannot_admit(transition: Transition) { + let f = Fixture::new().await; + let accepting = &f.state.workflow_engine; + let stale = Arc::new(WorkflowEngine::new( + f.state.db.clone(), + WorkflowConfig::default(), + )); + let accepting_sink = Arc::new(RecordingSink::default()); + let stale_sink = Arc::new(RecordingSink::default()); + accepting.set_action_sink(accepting_sink.clone()); + stale.set_action_sink(stale_sink.clone()); + assert!( + f.send(&definition(&f, f.now, true, "old-action")) + .await + .expect("create") + .accepted + ); + let selected = f + .state + .db + .get_workflow(f.tenant.community(), f.id) + .await + .expect("selected revision"); + + // Prime both *real* caches without creating runs; message != reaction. + let warm = buzz_core::StoredEvent::new( + f.sign(Kind::Custom(9), f.now, "warm", vec![]), + Some(f.channel), + ); + accepting + .on_event(f.tenant.community(), &warm) + .await + .expect("warm accepting cache"); + let warmed_at = Instant::now(); + stale + .on_event(f.tenant.community(), &warm) + .await + .expect("warm stale cache"); + stale + .on_event(f.tenant.community(), &reaction(&f)) + .await + .expect("initial fire"); + completed(&f, 1).await; + assert_eq!(*stale_sink.0.lock().expect("sink"), ["old-action"]); + stale_sink.0.lock().expect("sink").clear(); + + let recreate = matches!( + transition, + Transition::Recreate | Transition::RecreateIdentical + ); + if recreate { + assert!(f.send(&f.delete(f.now + 1)).await.expect("delete").accepted); + assert!(runs(&f).await.is_empty(), "deletion cascades prior runs"); + } + let enabled = !matches!(transition, Transition::Disable); + let text = if matches!(transition, Transition::Update | Transition::Recreate) { + "new-action" + } else { + "old-action" + }; + assert!( + f.send(&definition(&f, f.now + 2, enabled, text)) + .await + .expect("acknowledged transition") + .accepted + ); + let current = f + .state + .db + .get_workflow(f.tenant.community(), f.id) + .await + .expect("current revision"); + assert_eq!(current.enabled, enabled); + if matches!(transition, Transition::RecreateIdentical) { + assert_eq!( + current.definition_hash, selected.definition_hash, + "exercise incarnation independently of hash" + ); + assert_ne!(current.created_at, selected.created_at); + } + + // No TTL sleep or cache injection: lifecycle acknowledgement orders the fire. + // Fail on a stalled host rather than falsely passing because moka expired. + assert!( + warmed_at.elapsed() < Duration::from_secs(5), + "fixture exceeded cache-freshness budget" + ); + stale + .on_event(f.tenant.community(), &reaction(&f)) + .await + .expect("stale pod fire attempt"); + assert!( + warmed_at.elapsed() < Duration::from_secs(10), + "cache TTL elapsed during probe" + ); + let prior_runs = if recreate { 0 } else { 1 }; + let after = runs(&f).await; + if after.len() > prior_runs { + completed(&f, after.len()).await; + } + assert_eq!( + after.len(), + prior_runs, + "{transition:?}: stale cache admitted a run; actions={:?}", + stale_sink.0.lock().expect("sink") + ); + assert!( + stale_sink.0.lock().expect("sink").is_empty(), + "stale action escaped" + ); + + // The accepting pod must use the current definition, not merely suppress all work. + accepting + .on_event(f.tenant.community(), &reaction(&f)) + .await + .expect("accepting pod fire"); + let expected = prior_runs + usize::from(enabled); + completed(&f, expected).await; + let actions = accepting_sink.0.lock().expect("sink"); + if enabled { + assert_eq!(*actions, [text]); + } else { + assert!(actions.is_empty()); + } +} + +#[tokio::test] +#[ignore = "requires PostgreSQL"] +async fn stale_engine_cannot_run_superseded_actions_after_update() { + stale_engine_cannot_admit(Transition::Update).await; +} +#[tokio::test] +#[ignore = "requires PostgreSQL"] +async fn stale_engine_cannot_run_after_acknowledged_disable() { + stale_engine_cannot_admit(Transition::Disable).await; +} +#[tokio::test] +#[ignore = "requires PostgreSQL"] +async fn stale_engine_cannot_bind_old_actions_to_recreated_uuid() { + stale_engine_cannot_admit(Transition::Recreate).await; +} +#[tokio::test] +#[ignore = "requires PostgreSQL"] +async fn stale_engine_cannot_bind_identical_definition_to_new_incarnation() { + stale_engine_cannot_admit(Transition::RecreateIdentical).await; +} + +/// Exercise the production scheduler loop without exposing a test-only tick API. +#[tokio::test] +#[ignore = "requires PostgreSQL"] +async fn cluster_global_scheduler_fences_selected_revision_after_writer_settles() { + let mut f = Fixture::new().await; + let engine = &f.state.workflow_engine; + let sink = Arc::new(RecordingSink::default()); + engine.set_action_sink(sink.clone()); + let make_schedule = |id: Uuid, text: &str| { + f.sign(Kind::Custom(KIND_WORKFLOW_DEF as u16), f.now, + &format!("name: scheduled fence\ntrigger:\n on: schedule\n cron: '* * * * * *'\nsteps:\n - id: emit\n action: send_message\n text: {text}\n"), + vec![vec!["d".into(), id.to_string()], vec!["h".into(), f.channel.to_string()]]) + }; + let stale_id = f.id; + assert!( + f.send(&make_schedule(stale_id, "stale-schedule")) + .await + .expect("stale workflow") + .accepted + ); + let control_id = Uuid::new_v4(); + assert!( + f.send(&make_schedule(control_id, "current-schedule")) + .await + .expect("control workflow") + .accepted + ); + let mut writer = f.pool.begin().await.expect("writer"); + let writer_pid: i32 = sqlx::query_scalar("SELECT pg_backend_pid()") + .fetch_one(&mut *writer) + .await + .expect("writer pid"); + // Ordinary MVCC scheduler selection still sees the old enabled revision; + // the admission fence must wait and recheck after this writer commits. + sqlx::query(r#"UPDATE workflows SET definition_hash = $3, definition = jsonb_set(definition, '{steps,0,text}', '"new-schedule"'::jsonb) WHERE community_id=$1 AND id=$2"#) + .bind(f.tenant.community().as_uuid()).bind(stale_id).bind(vec![0x99_u8;32]) + .execute(&mut *writer).await.expect("in-flight revision update"); + let task_engine = Arc::clone(engine); + let task = tokio::spawn(async move { task_engine.run().await }); + let wait = tokio::time::timeout(Duration::from_secs(75), async { + loop { + let blocked: bool = sqlx::query_scalar( + "SELECT EXISTS(SELECT 1 FROM pg_stat_activity WHERE datname = current_database() AND query LIKE '%WITH eligible AS%' AND $1 = ANY(pg_blocking_pids(pid)))") + .bind(writer_pid).fetch_one(&f.pool).await.expect("observe scheduler admission wait"); + if blocked { break; } + // If the caller refreshes and bypasses the fence, the loop reaches + // the newer control workflow; that is a failed control, not a timeout. + assert!(sink.0.lock().expect("sink").is_empty(), "scheduler ran before the revision writer settled"); + tokio::time::sleep(Duration::from_millis(20)).await; + } + }).await; + if wait.is_err() { + task.abort(); + } + wait.expect("scheduler must reach the real admission lock"); + writer.commit().await.expect("commit revision update"); + // created_at ordering puts the control after the stale candidate; its + // completed run proves the scheduler passed the rejected candidate. + f.id = control_id; + tokio::time::timeout(Duration::from_secs(5), async { + loop { + let rows = runs(&f).await; + if rows.len() == 1 && rows[0].status == RunStatus::Completed { + break; + } + tokio::time::sleep(Duration::from_millis(10)).await; + } + }) + .await + .expect("current schedule remains runnable"); + task.abort(); + let _ = task.await; + f.id = stale_id; + assert!( + runs(&f).await.is_empty(), + "scheduler admitted a superseded selection" + ); + assert_eq!(*sink.0.lock().expect("sink"), ["current-schedule"]); +} diff --git a/crates/buzz-relay/src/nip11.rs b/crates/buzz-relay/src/nip11.rs index e6b18cdd0f8..37754236912 100644 --- a/crates/buzz-relay/src/nip11.rs +++ b/crates/buzz-relay/src/nip11.rs @@ -43,6 +43,9 @@ pub struct RelayInfo { /// NIP-PL executor descriptor. Present only when push delivery is configured. #[serde(skip_serializing_if = "Option::is_none")] pub push: Option, + /// Forward workflow lifecycle contract, bound to this request host. + #[serde(skip_serializing_if = "Option::is_none")] + pub workflows: Option, /// URL of the relay software repository. pub software: String, /// Relay software version string. @@ -69,6 +72,16 @@ pub struct RelayInfo { pub relay_self: Option, } +/// Positive evidence of atomic definition/runtime saves and canonical forward deletion. +/// This does not reconcile historical split state or grant execution authority. +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)] +pub struct WorkflowDescriptor { + /// Lifecycle contract revision (not a software version or supported-kind list). + pub lifecycle: u32, + /// Normalized, resolved request host, including a non-default port. + pub host: String, +} + /// Public capability descriptor for relay-proxied GIF search. #[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)] pub struct GifDescriptor { @@ -208,6 +221,7 @@ impl RelayInfo { supported_nips, supported_extensions: Some(supported_extensions), push: None, + workflows: None, software: "https://github.com/block/buzz".to_string(), version: env!("CARGO_PKG_VERSION").to_string(), limitation: Some(relay_limitation(max_message_length)), @@ -293,7 +307,7 @@ pub(crate) async fn nip11_document(state: &crate::state::AppState, raw_host: &st admin_api.as_deref(), state.config.klipy.as_ref().map(|_| "klipy"), ); - let tenant_host = if state.config.push_enabled { + let tenant_host = if state.config.push_enabled || relay_self.is_some() { crate::tenant::bind_community(&state.db, raw_host) .await .ok() @@ -313,6 +327,16 @@ pub(crate) async fn nip11_document(state: &crate::state::AppState, raw_host: &st .push("nip-pl".to_string()); info.push = Some(push); } + if let (true, Some(_), Some(host)) = ( + state.config.advertise_workflow_lifecycle, + relay_self, + tenant_host, + ) { + info.supported_extensions + .get_or_insert_default() + .push("buzz-workflows".to_string()); + info.workflows = Some(WorkflowDescriptor { lifecycle: 1, host }); + } info } @@ -399,6 +423,9 @@ const _RELAY_INFO_BUILD_STATIC_INPUT_FENCE: fn( Option<&str>, ) -> RelayInfo = RelayInfo::build; +#[cfg(test)] +mod postgres_tests; + #[cfg(test)] mod tests { use super::*; diff --git a/crates/buzz-relay/src/nip11/postgres_tests.rs b/crates/buzz-relay/src/nip11/postgres_tests.rs new file mode 100644 index 00000000000..1c70e41da9f --- /dev/null +++ b/crates/buzz-relay/src/nip11/postgres_tests.rs @@ -0,0 +1,120 @@ +//! Exercise both served NIP-11 routes; nextest supplies an isolated database. +use std::sync::Arc; + +use axum::{body::Body, http::Request}; +use serde_json::{json, Value}; +use tower::ServiceExt; + +use crate::{config::Config, state::AppState}; + +async fn served_info(state: &Arc, path: &str, host: &str) -> Value { + let response = crate::router::build_router(state.clone()) + .oneshot( + Request::builder() + .uri(path) + .header("host", host) + .header("accept", "application/nostr+json") + .body(Body::empty()) + .expect("request"), + ) + .await + .expect("NIP-11 response"); + assert_eq!(response.status(), axum::http::StatusCode::OK); + let body = axum::body::to_bytes(response.into_body(), 64 * 1024) + .await + .expect("bounded NIP-11 body"); + serde_json::from_slice(&body).expect("NIP-11 JSON") +} + +#[tokio::test] +#[ignore = "requires isolated PostgreSQL"] +async fn workflow_advertisement_requires_activation_identity_and_host_on_both_routes() { + let url = std::env::var("BUZZ_TEST_DATABASE_URL").expect("isolated PostgreSQL URL"); + let pool = sqlx::PgPool::connect(&url).await.expect("database"); + let db = buzz_db::Db::from_pool(pool.clone()); + let host = format!("activation-{}.example:8443", uuid::Uuid::new_v4()); + db.ensure_configured_community(&host) + .await + .expect("community"); + let keys = nostr::Keys::generate(); + let mut config = Config::from_env().expect("config"); + config.database_url = url; + config.redis_url = "redis://127.0.0.1:1".into(); + config.admin = None; + // An unrelated enabled extension must neither bypass nor be hidden by + // lifecycle activation. No gateway, Redis, or relay process is contacted. + config.push_enabled = true; + let cache = tempfile::tempdir().expect("cache directory"); + config.git_pack_cache_path = cache.path().to_path_buf(); + let redis = deadpool_redis::Config::from_url(&config.redis_url) + .create_pool(Some(deadpool_redis::Runtime::Tokio1)) + .expect("redis pool"); + let pubsub = Arc::new( + buzz_pubsub::PubSubManager::new(&config.redis_url, redis.clone()) + .await + .expect("pubsub"), + ); + let engine = Arc::new(buzz_workflow::WorkflowEngine::new( + db.clone(), + buzz_workflow::WorkflowConfig::default(), + )); + let auth = buzz_auth::AuthService::new(config.auth.clone()); + let search = buzz_search::SearchService::new(pool.clone()); + let media = buzz_media::MediaStorage::new(&config.media).expect("media config"); + let (state, _) = AppState::new( + config, db, redis, None, pubsub, auth, search, engine, keys, media, + ); + let mut state = Arc::new(state); + + for enabled in [false, true] { + for stable in [false, true] { + let mutable = Arc::get_mut(&mut state).expect("sole state owner"); + let config = Arc::make_mut(&mut mutable.config); + config.advertise_workflow_lifecycle = enabled; + config.relay_private_key = + stable.then(|| mutable.relay_keypair.secret_key().to_secret_hex()); + for request_host in [ + host.to_uppercase(), + "unmapped.invalid".into(), + String::new(), + ] { + let bound = request_host.eq_ignore_ascii_case(&host); + for path in ["/", "/info"] { + let info = served_info(&state, path, &request_host).await; + let advertised = enabled && stable && bound; + assert_eq!( + info.get("workflows").is_some(), + advertised, + "{path} enabled={enabled} stable={stable} host={request_host}" + ); + let extensions = info["supported_extensions"].as_array().expect("extensions"); + assert_eq!(extensions.contains(&json!("buzz-workflows")), advertised); + if advertised { + assert_eq!(info["workflows"], json!({"lifecycle": 1, "host": host})); + } + assert_eq!( + info.get("self").is_some(), + stable, + "identity is independent of activation" + ); + assert_eq!( + info.get("push").is_some(), + bound, + "push is independent of activation" + ); + assert_eq!(extensions.contains(&json!("nip-pl")), bound); + } + } + } + } + // Enabled + stable still fails closed if host resolution fails at runtime. + pool.close().await; + for path in ["/", "/info"] { + let info = served_info(&state, path, &host).await; + assert!(info.get("workflows").is_none()); + assert!(!info["supported_extensions"] + .as_array() + .expect("extensions") + .contains(&json!("buzz-workflows"))); + } +} diff --git a/crates/buzz-relay/src/workflow_sink.rs b/crates/buzz-relay/src/workflow_sink.rs index 6450b15b282..77c472c625c 100644 --- a/crates/buzz-relay/src/workflow_sink.rs +++ b/crates/buzz-relay/src/workflow_sink.rs @@ -862,12 +862,6 @@ mod postgres_tests { ..Default::default() }; let trigger_ctx_json = serde_json::to_value(&trigger_ctx).expect("serialize trigger"); - let run_id = state - .db - .create_workflow_run(community, workflow_id, None, Some(&trigger_ctx_json)) - .await - .expect("create workflow run"); - // Load the definition back from Postgres before execution. This pins the // authority source to the durable owner-authored template rather than a // second test-only string passed directly to RelayActionSink. @@ -876,6 +870,13 @@ mod postgres_tests { .get_workflow(community, workflow_id) .await .expect("load stored workflow"); + let run_id = state + .db + .create_workflow_run(&stored_workflow, None, Some(&trigger_ctx_json)) + .await + .expect("create workflow run") + .expect("current workflow admitted"); + let stored_definition: buzz_workflow::WorkflowDef = serde_json::from_value(stored_workflow.definition).expect("parse stored definition"); let result = buzz_workflow::executor::execute_run( diff --git a/crates/buzz-workflow/src/lib.rs b/crates/buzz-workflow/src/lib.rs index ee1c7467762..15707e63951 100644 --- a/crates/buzz-workflow/src/lib.rs +++ b/crates/buzz-workflow/src/lib.rs @@ -91,15 +91,10 @@ pub struct WorkflowEngine { /// `(community_id, channel_id)`. Most channels have no workflows, so this /// removes one SELECT from nearly every ingested event. /// - /// Consistency: the relay invalidates this cache on its own pod at the two - /// workflow mutation sites (command upsert, NIP-09 deletion). There is - /// deliberately no cross-pod invalidation — workflow triggering is not an - /// access-control fence, so the worst case on another pod is a just-deleted - /// workflow firing (or a just-created one missing events) for up to the TTL. - /// The same TTL also bounds the same-pod look-aside race (a stale fill - /// landing just after an invalidation). Workflow mutations are rare; the - /// 10s window matches the relay's other moka caches (see `AppState` in - /// `buzz-relay`). + /// Consistency: mutations invalidate the accepting pod's cache. Other pods + /// (and stale look-aside fills) may miss new triggers until the 10s TTL, + /// but cached records cannot admit superseded, disabled or recreated work: + /// run creation atomically fences the selected record against the live row. pub(crate) workflow_cache: moka::sync::Cache<(CommunityId, Uuid), Arc>>, } @@ -405,14 +400,17 @@ impl WorkflowEngine { let run_id = match self .db .create_workflow_run( - community_id, - workflow.id, + workflow, Some(&trigger_event_id_bytes), Some(&trigger_ctx_json), ) .await { - Ok(id) => id, + Ok(Some(id)) => id, + Ok(None) => { + tracing::debug!(workflow_id = %workflow.id, "Skipping stale or inactive workflow"); + continue; + } Err(e) => { tracing::error!(workflow_id = %workflow.id, "Failed to create run: {e}"); continue; @@ -667,14 +665,17 @@ impl WorkflowEngine { let run_id = match self .db .create_workflow_run( - community_id, - workflow.id, + workflow, None, // no trigger event for cron trigger_ctx_json.as_ref(), ) .await { - Ok(id) => id, + Ok(Some(id)) => id, + Ok(None) => { + tracing::debug!(workflow_id = %workflow.id, "Cron tick: skipping stale or inactive workflow"); + continue; + } Err(e) => { tracing::error!( workflow_id = %workflow.id, diff --git a/deploy/charts/buzz/README.md b/deploy/charts/buzz/README.md index 5e778279130..bfd02817c7c 100644 --- a/deploy/charts/buzz/README.md +++ b/deploy/charts/buzz/README.md @@ -291,12 +291,60 @@ default so long-lived WebSocket connections have time to drain. ## Upgrades -Schema migrations are embedded in the relay binary via `sqlx::migrate!` and run at startup, gated by `BUZZ_AUTO_MIGRATE` (default `true`). Multiple replicas race-safely behind a Postgres advisory lock. `helm upgrade` is the entire upgrade procedure. +Schema migrations are embedded in the relay binary via `sqlx::migrate!` and run at startup, gated by `BUZZ_AUTO_MIGRATE` (default `true`). Multiple replicas race-safely behind a Postgres advisory lock. Use `helm upgrade` for code rollouts; host-wide capability activation may require a separate post-drain step, as described below. Migration 0032 is a hard compatibility boundary for relay versions that publish repaired channel rosters. The relay verifies the roster-fence trigger catalog and behavior before opening listeners and refuses to start if 0032 is missing or inert. Apply migrations before rolling the relay; for large installations, prefer a controlled `buzz-admin migrate` job with PostgreSQL lock monitoring before the code rollout. If you prefer decoupling migrations from serving, set `migrate.autoMigrate=false`. **In that mode the chart does not run migrations for you** — you own running `buzz-admin migrate` (separate Pod / one-shot Job) against the database before every `helm install` / `helm upgrade`. Readiness probes only verify DB connectivity, not schema freshness, so a pod will appear healthy against an unmigrated schema and fail under load. A pre-upgrade Helm Job for this is on the chart roadmap; the values knob `migrate.preUpgradeJob.enabled` is reserved. +### Workflow lifecycle activation + +The NIP-11 `buzz-workflows` extension and `workflows: {lifecycle: 1, host: ...}` +are a **host-wide** forward save/delete guarantee, not a statement about the pod +that answered discovery. `BUZZ_ADVERTISE_WORKFLOW_LIFECYCLE` defaults to off; +only `true` or `1` enables it. This startup setting gates advertisement only: +new binaries enforce atomic lifecycle writes even while it is off. Stable relay +identity and successful request-host binding are still required to advertise. + +Do not enable it in the same rolling upgrade that first installs lifecycle-capable +code. `RollingUpdate` and the version-independent Service can route a later +connection to an old replica; existing WebSockets can also continue writing +through an old replica after it leaves the Service endpoints. The additive +`workflow_deletions` migration/serving fence does not drain those old processes. + +1. **Deploy with advertisement off** (unset or `false` on every replica). Apply + the required schema and roll out the lifecycle-capable image. While old and + new replicas coexist, clients must not rely on lifecycle 1. Existing split + state, including writes made during this phase, is not repaired by activation. +2. **Verify drain before enabling.** Wait for rollout completion, then verify + every HTTP/WebSocket endpoint behind every served host uses compatible code. + Include canary pools, alternate ingress routes and other deployments sharing + those hosts. Remove incompatible endpoints and finish draining/terminating + their existing connections and in-flight writes. A single successful NIP-11 + probe or a ready new pod is not evidence of this host-wide condition. +3. **Activate separately**, retaining the compatible image, using the existing + `relay.extraEnv` list (append this entry; preserve other entries): + + ```yaml + relay: + extraEnv: + - name: BUZZ_ADVERTISE_WORKFLOW_LIFECYCLE + value: "true" + ``` + + Run a second `helm upgrade`/GitOps sync and wait for this config rollout. + Mixed off/on replicas are now safe because all replicas already enforce the + contract. Verify `/info` and `/` with `Accept: application/nostr+json` for each + served host: `buzz-workflows` is listed, `workflows.lifecycle` is `1`, and + `workflows.host` matches the normalized request host. + +**Rollback:** switching this flag off does not disable enforcement or revoke +capabilities clients already observed. Prefer rolling back only to compatible +code. Before reintroducing incompatible code, stop traffic for the affected +hosts, disable advertisement everywhere, drain active connections/in-flight +writes, and require clients to discard cached discovery and rediscover before +resuming. If that cannot be guaranteed, do not downgrade those hosts. + ## Backups Save these. Losing any of them is data loss. See NOTES.txt printed by `helm install` for the live list: diff --git a/migrations/0045_workflow_deletion_cutoff.sql b/migrations/0045_workflow_deletion_cutoff.sql new file mode 100644 index 00000000000..cad6f6b9889 --- /dev/null +++ b/migrations/0045_workflow_deletion_cutoff.sql @@ -0,0 +1,13 @@ +-- Forward-only proof of atomic canonical workflow deletion. Never backfill from +-- legacy kind-5 events: their side effects may not have committed. +CREATE TABLE workflow_deletions ( + community_id UUID NOT NULL REFERENCES communities(id), + owner_pubkey BYTEA NOT NULL CHECK (octet_length(owner_pubkey) = 32), + workflow_id UUID NOT NULL, + channel_id UUID NOT NULL, + deleted_through TIMESTAMPTZ NOT NULL, + event_id BYTEA NOT NULL CHECK (octet_length(event_id) = 32), + PRIMARY KEY (community_id, owner_pubkey, workflow_id), + FOREIGN KEY (community_id, channel_id) REFERENCES channels(community_id, id) +); +SELECT attach_community_write_fence('workflow_deletions'); diff --git a/schema/schema.sql b/schema/schema.sql index 09508125622..a5f108e3483 100644 --- a/schema/schema.sql +++ b/schema/schema.sql @@ -382,6 +382,21 @@ CREATE INDEX idx_workflows_channel_active ON workflows (community_id, channel_id -- side effects run under the owning tenant's context (Lane0 contract §4a.5). CREATE INDEX idx_workflows_enabled ON workflows (enabled, status) WHERE enabled; +-- ── Workflow deletion cutoff ────────────────────────────────────────────────── + +-- Forward-only proof of atomic canonical workflow deletion. Never backfill from +-- legacy kind-5 events: their side effects may not have committed. +CREATE TABLE workflow_deletions ( + community_id UUID NOT NULL REFERENCES communities(id), + owner_pubkey BYTEA NOT NULL CHECK (octet_length(owner_pubkey) = 32), + workflow_id UUID NOT NULL, + channel_id UUID NOT NULL, + deleted_through TIMESTAMPTZ NOT NULL, + event_id BYTEA NOT NULL CHECK (octet_length(event_id) = 32), + PRIMARY KEY (community_id, owner_pubkey, workflow_id), + FOREIGN KEY (community_id, channel_id) REFERENCES channels(community_id, id) +); + -- ── Workflow runs ───────────────────────────────────────────────────────────── CREATE TABLE workflow_runs ( @@ -1752,6 +1767,7 @@ SELECT attach_community_write_fence('thread_metadata'); SELECT attach_community_write_fence('users'); SELECT attach_community_write_fence('workflow_approvals'); SELECT attach_community_write_fence('workflow_runs'); +SELECT attach_community_write_fence('workflow_deletions'); SELECT attach_community_write_fence('workflows'); -- ── Relay operator/moderator roster ──────────────────────────────────────────