diff --git a/crates/buzz-db/src/store/workflow.rs b/crates/buzz-db/src/store/workflow.rs index 68ae7a7fe4..f3e3b59ba0 100644 --- a/crates/buzz-db/src/store/workflow.rs +++ b/crates/buzz-db/src/store/workflow.rs @@ -10,6 +10,9 @@ /// 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; @@ -867,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, @@ -1343,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, ) @@ -2452,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) @@ -2481,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 @@ -2762,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/run_admission.rs b/crates/buzz-db/src/store/workflow/run_admission.rs new file mode 100644 index 0000000000..9777c30180 --- /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 37c549610d..664bf7ed82 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/handlers/command_executor.rs b/crates/buzz-relay/src/handlers/command_executor.rs index 09aad9d617..96d2c8e357 100644 --- a/crates/buzz-relay/src/handlers/command_executor.rs +++ b/crates/buzz-relay/src/handlers/command_executor.rs @@ -953,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() @@ -1675,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 0000000000..7fd09d08bb --- /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/workflow_sink.rs b/crates/buzz-relay/src/workflow_sink.rs index 6450b15b28..77c472c625 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 ee1c746776..15707e6395 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,