From b457d615652cdc64cce1a822f5a6bb19c718e8f2 Mon Sep 17 00:00:00 2001 From: Nicholas Craig Date: Tue, 28 Jul 2026 23:18:15 +0000 Subject: [PATCH] feat(workflow): make execution recovery clippy-clean Signed-off-by: Nicholas Craig --- Cargo.lock | 1 + crates/buzz-db/src/event.rs | 2 +- crates/buzz-db/src/lib.rs | 242 ++++ crates/buzz-db/src/migration.rs | 4 +- crates/buzz-db/src/workflow.rs | 1038 ++++++++++++++++- .../src/handlers/command_executor.rs | 21 +- crates/buzz-relay/src/workflow_sink.rs | 46 +- crates/buzz-workflow/Cargo.toml | 1 + crates/buzz-workflow/src/action_sink.rs | 2 + crates/buzz-workflow/src/executor.rs | 300 ++++- crates/buzz-workflow/src/lib.rs | 393 +++++-- crates/buzz-workflow/src/recovery.rs | 152 +++ migrations/0026_workflow_run_leases.sql | 24 + migrations/0027_workflow_step_attempts.sql | 33 + migrations/0028_workflow_event_outbox.sql | 41 + schema/schema.sql | 82 ++ 16 files changed, 2194 insertions(+), 188 deletions(-) create mode 100644 crates/buzz-workflow/src/recovery.rs create mode 100644 migrations/0026_workflow_run_leases.sql create mode 100644 migrations/0027_workflow_step_attempts.sql create mode 100644 migrations/0028_workflow_event_outbox.sql diff --git a/Cargo.lock b/Cargo.lock index 3b60dc4579..d05dff83d0 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1273,6 +1273,7 @@ dependencies = [ "dashmap", "evalexpr", "hex", + "metrics", "moka", "nostr", "reqwest 0.13.4", diff --git a/crates/buzz-db/src/event.rs b/crates/buzz-db/src/event.rs index 520fd1536f..96ead333dd 100644 --- a/crates/buzz-db/src/event.rs +++ b/crates/buzz-db/src/event.rs @@ -1042,7 +1042,7 @@ pub struct ThreadMetadataParams<'a> { pub broadcast: bool, } -async fn insert_event_with_thread_metadata_tx( +pub(crate) async fn insert_event_with_thread_metadata_tx( tx: &mut Transaction<'_, Postgres>, community_id: CommunityId, event: &Event, diff --git a/crates/buzz-db/src/lib.rs b/crates/buzz-db/src/lib.rs index 2a3ba9a63e..df8cf81854 100644 --- a/crates/buzz-db/src/lib.rs +++ b/crates/buzz-db/src/lib.rs @@ -1401,6 +1401,51 @@ impl Db { Ok(result) } + /// Atomically persist a workflow event and its action-effect receipt. + /// + /// This closes the crash window between event publication and recording + /// the event ID used for idempotent recovery. + #[allow(clippy::too_many_arguments)] + pub async fn insert_workflow_action_event( + &self, + community_id: CommunityId, + run_id: Uuid, + idempotency_key: &str, + event: &nostr::Event, + channel_id: Option, + thread_meta: Option>, + ) -> Result<(StoredEvent, bool)> { + let mut tx = self.pool.begin().await?; + let result = event::insert_event_with_thread_metadata_tx( + &mut tx, + community_id, + event, + channel_id, + thread_meta, + ) + .await?; + let recorded = workflow::complete_workflow_action_effect_on_connection( + &mut tx, + community_id, + run_id, + idempotency_key, + event.id.as_bytes(), + ) + .await?; + if !recorded { + return Err(DbError::InvalidData( + "workflow action effect reservation was lost before event commit".into(), + )); + } + tx.commit().await?; + if result.1 { + if let Err(e) = insert_mentions(&self.pool, community_id, event, channel_id).await { + tracing::warn!(event_id = %event.id, "Failed to insert mentions: {e}"); + } + } + Ok(result) + } + /// Atomically insert a kind:7 reaction event and its reaction row. #[allow(clippy::too_many_arguments)] pub async fn insert_reaction_event_with_thread_metadata( @@ -2566,6 +2611,24 @@ impl Db { .await } + /// Atomically claim a scheduled fire and create/attach its workflow run. + pub async fn claim_and_create_scheduled_workflow_run( + &self, + community_id: CommunityId, + workflow_id: Uuid, + scheduled_for: chrono::DateTime, + trigger_context: Option<&serde_json::Value>, + ) -> Result> { + workflow::claim_and_create_scheduled_workflow_run( + &self.pool, + community_id, + workflow_id, + scheduled_for, + trigger_context, + ) + .await + } + /// Fetch the latest claimed schedule instant for interval trigger anchoring. pub async fn latest_scheduled_workflow_fire( &self, @@ -2744,6 +2807,185 @@ impl Db { .await } + /// Claim one runnable workflow run with a durable fencing lease. + pub async fn claim_workflow_run( + &self, + community_id: CommunityId, + owner: &str, + lease_for_secs: i64, + ) -> Result> { + workflow::claim_workflow_run(&self.pool, community_id, owner, lease_for_secs).await + } + + /// Claim a specific workflow run with a fencing lease. + pub async fn claim_specific_workflow_run( + &self, + community_id: CommunityId, + run_id: Uuid, + owner: &str, + lease_for_secs: i64, + ) -> Result> { + workflow::claim_specific_workflow_run( + &self.pool, + community_id, + run_id, + owner, + lease_for_secs, + ) + .await + } + + /// Claim a run only after its approval gate has been explicitly granted. + pub async fn claim_waiting_approval_workflow_run( + &self, + community_id: CommunityId, + run_id: Uuid, + owner: &str, + lease_for_secs: i64, + ) -> Result> { + workflow::claim_waiting_approval_workflow_run( + &self.pool, + community_id, + run_id, + owner, + lease_for_secs, + ) + .await + } + + /// Heartbeat a workflow lease using owner and fencing token CAS. + pub async fn heartbeat_workflow_run( + &self, + lease: &workflow::WorkflowRunLease, + lease_for_secs: i64, + ) -> Result { + workflow::heartbeat_workflow_run(&self.pool, lease, lease_for_secs).await + } + + /// Finalize a workflow run only while its fencing lease remains valid. + pub async fn finalize_workflow_run( + &self, + lease: &workflow::WorkflowRunLease, + status: workflow::RunStatus, + current_step: i32, + trace: &serde_json::Value, + error: Option<&str>, + recovery_classification: Option<&str>, + ) -> Result { + workflow::finalize_workflow_run( + &self.pool, + lease, + status, + current_step, + trace, + error, + recovery_classification, + ) + .await + } + + /// Release a workflow run until its durable delay deadline. + pub async fn defer_workflow_run( + &self, + lease: &workflow::WorkflowRunLease, + current_step: i32, + trace: &serde_json::Value, + available_at: chrono::DateTime, + ) -> Result { + workflow::defer_workflow_run(&self.pool, lease, current_step, trace, available_at).await + } + + /// Persist the start of a workflow step attempt. + pub async fn start_workflow_step_attempt( + &self, + community_id: CommunityId, + run_id: Uuid, + step_index: i32, + step_id: &str, + ) -> Result { + workflow::start_workflow_step_attempt(&self.pool, community_id, run_id, step_index, step_id) + .await + } + + /// Persist a terminal workflow step attempt result. + #[allow(clippy::too_many_arguments)] + pub async fn complete_workflow_step_attempt( + &self, + community_id: CommunityId, + run_id: Uuid, + step_index: i32, + attempt: i32, + status: &str, + result: Option<&serde_json::Value>, + error: Option<&str>, + recovery_classification: Option<&str>, + ) -> Result { + workflow::complete_workflow_step_attempt( + &self.pool, + community_id, + run_id, + step_index, + attempt, + status, + result, + error, + recovery_classification, + ) + .await + } + + /// Reserve a workflow remote side effect by stable idempotency key. + pub async fn reserve_workflow_action_effect( + &self, + community_id: CommunityId, + run_id: Uuid, + idempotency_key: &str, + ) -> Result { + workflow::reserve_workflow_action_effect(&self.pool, community_id, run_id, idempotency_key) + .await + } + + /// Persist the remote event identity for a reserved effect. + pub async fn complete_workflow_action_effect( + &self, + community_id: CommunityId, + idempotency_key: &str, + event_id: &[u8], + ) -> Result { + workflow::complete_workflow_action_effect( + &self.pool, + community_id, + idempotency_key, + event_id, + ) + .await + } + + /// Claim durable event-trigger handoffs. + pub async fn claim_workflow_event_outbox( + &self, + owner: &str, + limit: i64, + ) -> Result> { + workflow::claim_workflow_event_outbox(&self.pool, owner, limit).await + } + + /// Acknowledge a processed event-trigger handoff. + pub async fn ack_workflow_event_outbox( + &self, + item: &workflow::WorkflowEventOutboxItem, + ) -> Result { + workflow::ack_workflow_event_outbox(&self.pool, item).await + } + + /// Release a failed event-trigger handoff for retry. + pub async fn release_workflow_event_outbox( + &self, + item: &workflow::WorkflowEventOutboxItem, + ) -> Result { + workflow::release_workflow_event_outbox(&self.pool, item).await + } + /// Create an approval request. pub async fn create_approval(&self, params: workflow::CreateApprovalParams<'_>) -> Result<()> { workflow::create_approval(&self.pool, params).await diff --git a/crates/buzz-db/src/migration.rs b/crates/buzz-db/src/migration.rs index 1d1b7e05d4..96a13d78db 100644 --- a/crates/buzz-db/src/migration.rs +++ b/crates/buzz-db/src/migration.rs @@ -560,7 +560,7 @@ mod tests { let mut migrations: Vec<_> = MIGRATOR.iter().collect(); migrations.sort_by_key(|migration| migration.version); - assert_eq!(migrations.len(), 25); + assert_eq!(migrations.len(), 28); assert_eq!(migrations[0].version, 1); assert_eq!(&*migrations[0].description, "initial schema"); assert!(migrations[0] @@ -1146,7 +1146,7 @@ mod tests { run_migrations(&pool) .await .expect("retry succeeds after operator repair"); - assert_eq!(applied_versions(&pool).await.last().copied(), Some(25)); + assert_eq!(applied_versions(&pool).await.last().copied(), Some(28)); } #[tokio::test] diff --git a/crates/buzz-db/src/workflow.rs b/crates/buzz-db/src/workflow.rs index 7a2396c1fd..3ccb589402 100644 --- a/crates/buzz-db/src/workflow.rs +++ b/crates/buzz-db/src/workflow.rs @@ -12,7 +12,7 @@ use std::str::FromStr; use chrono::{DateTime, Utc}; use sha2::{Digest, Sha256}; -use sqlx::{PgPool, Row}; +use sqlx::{PgConnection, PgPool, Row}; use uuid::Uuid; use buzz_core::CommunityId; @@ -220,6 +220,64 @@ pub struct WorkflowRunRecord { pub error_message: Option, /// When the run record was created. pub created_at: DateTime, + /// Current lease owner, when a worker owns this run. + pub lease_owner: Option, + /// Current fencing token, when a worker owns this run. + pub lease_token: Option, + /// Lease expiry, if currently leased. + pub lease_expires_at: Option>, + /// Number of claim attempts. + pub attempt: i32, + /// Earliest time this run may be claimed. + pub available_at: DateTime, + /// Whether a failure is eligible for retry or terminal. + pub recovery_classification: Option, +} + +/// A worker's durable lease and fencing token. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct WorkflowRunLease { + /// Run identity. + pub run_id: Uuid, + /// Community owning the run. + pub community_id: CommunityId, + /// Worker identity that owns the lease. + pub owner: String, + /// Current fencing token. + pub token: Uuid, + /// Attempt number assigned by the claim. + pub attempt: i32, + /// Lease expiry. + pub expires_at: DateTime, +} + +/// Durable identity assigned before a step side effect begins. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct WorkflowStepAttempt { + /// Attempt number for this step. + pub attempt: i32, + /// Stable side-effect deduplication key. + pub idempotency_key: String, +} + +/// Existing or newly reserved remote action effect. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct WorkflowActionEffect { + /// Whether this caller created the reservation. + pub newly_reserved: bool, + /// Previously persisted remote event identity, if known. + pub event_id: Option>, +} + +/// A claimed event-trigger handoff from the durable outbox. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct WorkflowEventOutboxItem { + /// Owning tenant. + pub community_id: CommunityId, + /// Persisted event identity. + pub event_id: Vec, + /// Claim owner used for ack/release fencing. + pub claimed_by: String, } /// A winning scheduled workflow fire claim. @@ -239,6 +297,16 @@ pub struct ScheduledWorkflowFireClaim { pub claimed_at: DateTime, } +/// A scheduled fire claim whose workflow run was created and attached in the +/// same database transaction. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct ScheduledWorkflowRunClaim { + /// The durable fire identity that won the claim. + pub fire: ScheduledWorkflowFireClaim, + /// The run created for the claimed fire. + pub run_id: Uuid, +} + /// A pending or resolved approval gate for a workflow step. #[derive(Debug, Clone)] pub struct ApprovalRecord { @@ -587,6 +655,96 @@ pub async fn attach_scheduled_workflow_run( Ok(result.rows_affected() == 1) } +/// Atomically claim a scheduled fire, create its pending run, and attach the +/// run to the claim. A failed run insert or attach rolls back the fire claim, +/// allowing a later scheduler tick to retry instead of leaving an orphaned +/// at-most-once claim. +pub async fn claim_and_create_scheduled_workflow_run( + pool: &PgPool, + community_id: CommunityId, + workflow_id: Uuid, + scheduled_for: DateTime, + trigger_context: Option<&serde_json::Value>, +) -> Result> { + let mut tx = pool.begin().await?; + let row = sqlx::query( + r#" + INSERT INTO scheduled_workflow_fires (community_id, workflow_id, scheduled_for) + SELECT w.community_id, w.id, $3 + FROM workflows w + WHERE w.community_id = $1 AND w.id = $2 + ON CONFLICT (community_id, workflow_id, scheduled_for) DO NOTHING + RETURNING community_id, workflow_id, scheduled_for, claimed_at + "#, + ) + .bind(community_id.as_uuid()) + .bind(workflow_id) + .bind(scheduled_for) + .fetch_optional(&mut *tx) + .await?; + + let Some(row) = row else { + tx.commit().await?; + return Ok(None); + }; + + let run_id = Uuid::new_v4(); + let inserted = sqlx::query( + r#" + INSERT INTO workflow_runs + (community_id, id, workflow_id, definition_snapshot, definition_hash, status, trigger_event_id, current_step, execution_trace, trigger_context) + SELECT $1, $2, w.id, w.definition, w.definition_hash, 'pending', NULL, 0, '[]', $4 + FROM workflows w WHERE w.community_id = $1 AND w.id = $3 + ON CONFLICT (community_id, workflow_id, trigger_event_id) + WHERE trigger_event_id IS NOT NULL DO NOTHING + RETURNING id + "#, + ) + .bind(community_id.as_uuid()) + .bind(run_id) + .bind(workflow_id) + .bind(trigger_context) + .execute(&mut *tx) + .await?; + if inserted.rows_affected() != 1 { + return Err(DbError::NotFound(format!("workflow {workflow_id}"))); + } + + let attached = sqlx::query( + r#" + UPDATE scheduled_workflow_fires + SET workflow_run_id = $4 + WHERE community_id = $1 + AND workflow_id = $2 + AND scheduled_for = $3 + AND workflow_run_id IS NULL + "#, + ) + .bind(community_id.as_uuid()) + .bind(workflow_id) + .bind(scheduled_for) + .bind(run_id) + .execute(&mut *tx) + .await? + .rows_affected(); + if attached != 1 { + return Err(DbError::InvalidData( + "scheduled workflow run claim lost its attachment".into(), + )); + } + + tx.commit().await?; + Ok(Some(ScheduledWorkflowRunClaim { + fire: ScheduledWorkflowFireClaim { + community_id: CommunityId::from_uuid(row.try_get("community_id")?), + workflow_id: row.try_get("workflow_id")?, + scheduled_for: row.try_get("scheduled_for")?, + claimed_at: row.try_get("claimed_at")?, + }, + run_id, + })) +} + /// Delete old scheduled workflow fire claims for retention. /// /// Schedule claim rows are correctness metadata, but they grow with every fire. @@ -803,11 +961,15 @@ pub async fn create_workflow_run( ) -> Result { let id = Uuid::new_v4(); - sqlx::query( + let inserted = 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) + (community_id, id, workflow_id, definition_snapshot, definition_hash, status, trigger_event_id, current_step, execution_trace, trigger_context) + SELECT $1, $2, w.id, w.definition, w.definition_hash, 'pending', $4, 0, '[]', $5 + FROM workflows w WHERE w.community_id = $1 AND w.id = $3 + ON CONFLICT (community_id, workflow_id, trigger_event_id) + WHERE trigger_event_id IS NOT NULL DO NOTHING + RETURNING id "#, ) .bind(community_id.as_uuid()) @@ -815,10 +977,57 @@ pub async fn create_workflow_run( .bind(workflow_id) .bind(trigger_event_id) .bind(trigger_context) - .execute(pool) + .fetch_optional(pool) .await?; - Ok(id) + if let Some(row) = inserted { + return Ok(row.try_get("id")?); + } + sqlx::query_scalar( + "SELECT id FROM workflow_runs WHERE community_id = $1 AND workflow_id = $2 AND trigger_event_id = $3", + ) + .bind(community_id.as_uuid()) + .bind(workflow_id) + .bind(trigger_event_id) + .fetch_optional(pool) + .await? + .ok_or_else(|| DbError::NotFound(format!("workflow {workflow_id}"))) +} + +/// Create a workflow run on the caller's open transaction. Used by command +/// ingress so the trigger event and its run commit or roll back together. +pub async fn create_workflow_run_on_connection( + conn: &mut PgConnection, + community_id: CommunityId, + workflow_id: Uuid, + trigger_event_id: Option<&[u8]>, + trigger_context: Option<&serde_json::Value>, +) -> Result { + let id = Uuid::new_v4(); + let inserted = sqlx::query( + r#" + INSERT INTO workflow_runs + (community_id, id, workflow_id, definition_snapshot, definition_hash, status, trigger_event_id, current_step, execution_trace, trigger_context) + SELECT $1, $2, w.id, w.definition, w.definition_hash, 'pending', $4, 0, '[]', $5 + FROM workflows w WHERE w.community_id = $1 AND w.id = $3 + ON CONFLICT (community_id, workflow_id, trigger_event_id) + WHERE trigger_event_id IS NOT NULL DO NOTHING + RETURNING id + "#, + ) + .bind(community_id.as_uuid()) + .bind(id) + .bind(workflow_id) + .bind(trigger_event_id) + .bind(trigger_context) + .fetch_optional(&mut *conn) + .await?; + inserted + .map(|row| row.try_get("id")) + .transpose()? + .ok_or_else(|| { + DbError::InvalidData("workflow trigger already has a run or workflow is missing".into()) + }) } /// Fetch a single workflow run by ID, scoped to its community. @@ -830,7 +1039,8 @@ pub async fn get_workflow_run( let row = sqlx::query( r#" SELECT community_id, id, workflow_id, status::text AS status, trigger_event_id, current_step, - execution_trace, trigger_context, started_at, completed_at, error_message, created_at + execution_trace, trigger_context, started_at, completed_at, error_message, created_at, + lease_owner, lease_token, lease_expires_at, attempt, available_at, recovery_classification FROM workflow_runs WHERE community_id = $1 AND id = $2 "#, @@ -855,7 +1065,8 @@ pub async fn list_workflow_runs( let rows = sqlx::query( r#" SELECT community_id, id, workflow_id, status::text AS status, trigger_event_id, current_step, - execution_trace, trigger_context, started_at, completed_at, error_message, created_at + execution_trace, trigger_context, started_at, completed_at, error_message, created_at, + lease_owner, lease_token, lease_expires_at, attempt, available_at, recovery_classification FROM workflow_runs WHERE community_id = $1 AND workflow_id = $2 ORDER BY created_at DESC @@ -919,6 +1130,466 @@ pub async fn update_workflow_run( Ok(()) } +/// Claim the oldest runnable workflow run with a new fencing token. +pub async fn claim_workflow_run( + pool: &PgPool, + community_id: CommunityId, + owner: &str, + lease_for_secs: i64, +) -> Result> { + let row = sqlx::query( + r#" + WITH candidate AS ( + SELECT id + FROM workflow_runs + WHERE community_id = $1 + AND available_at <= NOW() + AND ( + status = 'pending' + OR (status = 'failed' AND recovery_classification = 'retryable') + OR (status = 'running' AND lease_expires_at < NOW()) + ) + ORDER BY available_at, created_at + FOR UPDATE SKIP LOCKED + LIMIT 1 + ) + UPDATE workflow_runs r + SET status = 'running', + lease_owner = $2, + lease_token = gen_random_uuid(), + lease_expires_at = NOW() + ($3 * INTERVAL '1 second'), + heartbeat_at = NOW(), + attempt = r.attempt + 1, + started_at = COALESCE(r.started_at, NOW()), + recovery_classification = NULL + FROM candidate c + WHERE r.community_id = $1 AND r.id = c.id + RETURNING r.id, r.community_id, r.lease_token, r.attempt, r.lease_expires_at + "#, + ) + .bind(community_id.as_uuid()) + .bind(owner) + .bind(lease_for_secs) + .fetch_optional(pool) + .await?; + + row.map(|row| { + Ok(WorkflowRunLease { + run_id: row.try_get("id")?, + community_id: CommunityId::from_uuid(row.try_get("community_id")?), + owner: owner.to_owned(), + token: row.try_get("lease_token")?, + attempt: row.try_get("attempt")?, + expires_at: row.try_get("lease_expires_at")?, + }) + }) + .transpose() +} + +/// Claim a specific run when an ingress path has already selected its ID. +pub async fn claim_specific_workflow_run( + pool: &PgPool, + community_id: CommunityId, + run_id: Uuid, + owner: &str, + lease_for_secs: i64, +) -> Result> { + let row = sqlx::query( + r#" + UPDATE workflow_runs + SET status = 'running', lease_owner = $3, lease_token = gen_random_uuid(), + lease_expires_at = NOW() + ($4 * INTERVAL '1 second'), heartbeat_at = NOW(), + attempt = attempt + 1, started_at = COALESCE(started_at, NOW()), + recovery_classification = NULL + WHERE community_id = $1 AND id = $2 AND available_at <= NOW() + AND (status = 'pending' + OR (status = 'failed' AND recovery_classification = 'retryable') + OR (status = 'running' AND lease_expires_at < NOW())) + RETURNING id, community_id, lease_token, attempt, lease_expires_at + "#, + ) + .bind(community_id.as_uuid()) + .bind(run_id) + .bind(owner) + .bind(lease_for_secs) + .fetch_optional(pool) + .await?; + row.map(|row| { + Ok(WorkflowRunLease { + run_id: row.try_get("id")?, + community_id: CommunityId::from_uuid(row.try_get("community_id")?), + owner: owner.to_owned(), + token: row.try_get("lease_token")?, + attempt: row.try_get("attempt")?, + expires_at: row.try_get("lease_expires_at")?, + }) + }) + .transpose() +} + +/// Claim a run that is resuming after an explicitly granted approval. +/// Generic scheduler claims intentionally exclude `waiting_approval` so a +/// background reconciliation cannot bypass the approval gate. +pub async fn claim_waiting_approval_workflow_run( + pool: &PgPool, + community_id: CommunityId, + run_id: Uuid, + owner: &str, + lease_for_secs: i64, +) -> Result> { + let row = sqlx::query( + r#" + UPDATE workflow_runs + SET status = 'running', lease_owner = $3, lease_token = gen_random_uuid(), + lease_expires_at = NOW() + ($4 * INTERVAL '1 second'), heartbeat_at = NOW(), + attempt = attempt + 1, started_at = COALESCE(started_at, NOW()), + recovery_classification = NULL + WHERE community_id = $1 AND id = $2 AND status = 'waiting_approval' + RETURNING id, community_id, lease_token, attempt, lease_expires_at + "#, + ) + .bind(community_id.as_uuid()) + .bind(run_id) + .bind(owner) + .bind(lease_for_secs) + .fetch_optional(pool) + .await?; + row.map(|row| { + Ok(WorkflowRunLease { + run_id: row.try_get("id")?, + community_id: CommunityId::from_uuid(row.try_get("community_id")?), + owner: owner.to_owned(), + token: row.try_get("lease_token")?, + attempt: row.try_get("attempt")?, + expires_at: row.try_get("lease_expires_at")?, + }) + }) + .transpose() +} + +/// Extend a lease only when its owner and fencing token still match. +pub async fn heartbeat_workflow_run( + pool: &PgPool, + lease: &WorkflowRunLease, + lease_for_secs: i64, +) -> Result { + let result = sqlx::query( + "UPDATE workflow_runs SET lease_expires_at = NOW() + ($1 * INTERVAL '1 second'), heartbeat_at = NOW() \ + WHERE community_id = $2 AND id = $3 AND status = 'running' \ + AND lease_owner = $4 AND lease_token = $5 AND lease_expires_at > NOW()", + ) + .bind(lease_for_secs) + .bind(lease.community_id.as_uuid()) + .bind(lease.run_id) + .bind(&lease.owner) + .bind(lease.token) + .execute(pool) + .await?; + Ok(result.rows_affected() == 1) +} + +/// Finalize a run only if this worker still owns its fencing token. +pub async fn finalize_workflow_run( + pool: &PgPool, + lease: &WorkflowRunLease, + status: RunStatus, + current_step: i32, + trace: &serde_json::Value, + error: Option<&str>, + recovery_classification: Option<&str>, +) -> Result { + let status_str = status.to_string(); + let result = sqlx::query( + r#" + UPDATE workflow_runs + SET status = $1::run_status, + current_step = $2, + execution_trace = $3, + error_message = $4, + recovery_classification = $5, + available_at = CASE WHEN $5 = 'retryable' THEN NOW() + INTERVAL '5 seconds' ELSE available_at END, + completed_at = CASE WHEN $1 IN ('completed','failed','cancelled') THEN NOW() ELSE completed_at END, + lease_owner = NULL, + lease_token = NULL, + lease_expires_at = NULL + WHERE community_id = $6 AND id = $7 + AND status = 'running' + AND lease_owner = $8 AND lease_token = $9 + AND lease_expires_at > NOW() + "#, + ) + .bind(&status_str) + .bind(current_step) + .bind(trace) + .bind(error) + .bind(recovery_classification) + .bind(lease.community_id.as_uuid()) + .bind(lease.run_id) + .bind(&lease.owner) + .bind(lease.token) + .execute(pool) + .await?; + Ok(result.rows_affected() == 1) +} + +/// Release a leased run for a durable delay without keeping a worker asleep. +pub async fn defer_workflow_run( + pool: &PgPool, + lease: &WorkflowRunLease, + current_step: i32, + trace: &serde_json::Value, + available_at: DateTime, +) -> Result { + let result = sqlx::query( + r#" + UPDATE workflow_runs + SET status = 'pending', current_step = $1, execution_trace = $2, + available_at = $3, lease_owner = NULL, lease_token = NULL, + lease_expires_at = NULL, heartbeat_at = NULL, + recovery_classification = 'delayed' + WHERE community_id = $4 AND id = $5 AND status = 'running' + AND lease_owner = $6 AND lease_token = $7 + AND lease_expires_at > NOW() + "#, + ) + .bind(current_step) + .bind(trace) + .bind(available_at) + .bind(lease.community_id.as_uuid()) + .bind(lease.run_id) + .bind(&lease.owner) + .bind(lease.token) + .execute(pool) + .await?; + Ok(result.rows_affected() == 1) +} + +/// Record the start of a step attempt before invoking its side effect. +pub async fn start_workflow_step_attempt( + pool: &PgPool, + community_id: CommunityId, + run_id: Uuid, + step_index: i32, + step_id: &str, +) -> Result { + if let Some(existing) = sqlx::query( + "SELECT attempt, idempotency_key FROM workflow_step_attempts \ + WHERE community_id = $1 AND run_id = $2 AND step_index = $3 \ + AND status = 'running' ORDER BY attempt DESC LIMIT 1", + ) + .bind(community_id.as_uuid()) + .bind(run_id) + .bind(step_index) + .fetch_optional(pool) + .await? + { + return Ok(WorkflowStepAttempt { + attempt: existing.try_get("attempt")?, + idempotency_key: existing.try_get("idempotency_key")?, + }); + } + + let row = sqlx::query( + r#" + WITH next AS ( + SELECT COALESCE(MAX(attempt), 0) + 1 AS attempt + FROM workflow_step_attempts + WHERE community_id = $1 AND run_id = $2 AND step_index = $3 + ) + INSERT INTO workflow_step_attempts + (community_id, run_id, step_index, step_id, attempt, idempotency_key, status) + SELECT $1, $2, $3, $4, attempt, + $2::text || ':' || $5::text || ':' || attempt::text, + 'running' + FROM next + RETURNING attempt, idempotency_key + "#, + ) + .bind(community_id.as_uuid()) + .bind(run_id) + .bind(step_index) + .bind(step_id) + .bind(step_id) + .fetch_one(pool) + .await?; + Ok(WorkflowStepAttempt { + attempt: row.try_get("attempt")?, + idempotency_key: row.try_get("idempotency_key")?, + }) +} + +/// Persist the terminal result of a step attempt. +#[allow(clippy::too_many_arguments)] +pub async fn complete_workflow_step_attempt( + pool: &PgPool, + community_id: CommunityId, + run_id: Uuid, + step_index: i32, + attempt: i32, + status: &str, + result: Option<&serde_json::Value>, + error: Option<&str>, + recovery_classification: Option<&str>, +) -> Result { + let changed = sqlx::query( + "UPDATE workflow_step_attempts SET status = $1, completed_at = NOW(), result = $2, \ + error_message = $3, recovery_classification = $4 \ + WHERE community_id = $5 AND run_id = $6 AND step_index = $7 AND attempt = $8 \ + AND status = 'running'", + ) + .bind(status) + .bind(result) + .bind(error) + .bind(recovery_classification) + .bind(community_id.as_uuid()) + .bind(run_id) + .bind(step_index) + .bind(attempt) + .execute(pool) + .await? + .rows_affected(); + Ok(changed == 1) +} + +/// Reserve a remote side effect before publication. +pub async fn reserve_workflow_action_effect( + pool: &PgPool, + community_id: CommunityId, + run_id: Uuid, + idempotency_key: &str, +) -> Result { + let row = sqlx::query( + r#" + INSERT INTO workflow_action_effects (community_id, run_id, idempotency_key) + VALUES ($1, $2, $3) + ON CONFLICT (community_id, idempotency_key) DO UPDATE + SET idempotency_key = EXCLUDED.idempotency_key + RETURNING (xmax = 0) AS newly_reserved, event_id + "#, + ) + .bind(community_id.as_uuid()) + .bind(run_id) + .bind(idempotency_key) + .fetch_one(pool) + .await?; + Ok(WorkflowActionEffect { + newly_reserved: row.try_get("newly_reserved")?, + event_id: row.try_get("event_id")?, + }) +} + +/// Persist the remote event identity after insertion. +pub async fn complete_workflow_action_effect( + pool: &PgPool, + community_id: CommunityId, + idempotency_key: &str, + event_id: &[u8], +) -> Result { + let result = sqlx::query( + "UPDATE workflow_action_effects SET event_id = $1, status = 'published' \ + WHERE community_id = $2 AND idempotency_key = $3 AND event_id IS NULL", + ) + .bind(event_id) + .bind(community_id.as_uuid()) + .bind(idempotency_key) + .execute(pool) + .await?; + Ok(result.rows_affected() == 1) +} + +/// Persist an action's outbound event identity using the caller's transaction. +pub async fn complete_workflow_action_effect_on_connection( + conn: &mut PgConnection, + community_id: CommunityId, + run_id: Uuid, + idempotency_key: &str, + event_id: &[u8], +) -> Result { + let result = sqlx::query( + "UPDATE workflow_action_effects SET event_id = $1, status = 'published' \ + WHERE community_id = $2 AND run_id = $3 AND idempotency_key = $4 AND event_id IS NULL", + ) + .bind(event_id) + .bind(community_id.as_uuid()) + .bind(run_id) + .bind(idempotency_key) + .execute(&mut *conn) + .await?; + Ok(result.rows_affected() == 1) +} + +/// Claim event-trigger handoffs durably; expired claims are reclaimable. +pub async fn claim_workflow_event_outbox( + pool: &PgPool, + owner: &str, + limit: i64, +) -> Result> { + let rows = sqlx::query( + r#" + WITH candidates AS ( + SELECT community_id, event_id + FROM workflow_event_outbox + WHERE available_at <= NOW() + AND (claimed_until IS NULL OR claimed_until < NOW()) + ORDER BY available_at, created_at + FOR UPDATE SKIP LOCKED + LIMIT $2 + ) + UPDATE workflow_event_outbox o + SET claimed_by = $1, claimed_until = NOW() + INTERVAL '60 seconds', attempts = attempts + 1 + FROM candidates c + WHERE o.community_id = c.community_id AND o.event_id = c.event_id + RETURNING o.community_id, o.event_id + "#, + ) + .bind(owner) + .bind(limit.clamp(1, 1000)) + .fetch_all(pool) + .await?; + rows.into_iter() + .map(|row| { + Ok(WorkflowEventOutboxItem { + community_id: CommunityId::from_uuid(row.try_get("community_id")?), + event_id: row.try_get("event_id")?, + claimed_by: owner.to_owned(), + }) + }) + .collect() +} + +/// Acknowledge a successfully processed event-trigger handoff. +pub async fn ack_workflow_event_outbox( + pool: &PgPool, + item: &WorkflowEventOutboxItem, +) -> Result { + let result = sqlx::query( + "DELETE FROM workflow_event_outbox WHERE community_id = $1 AND event_id = $2 AND claimed_by = $3", + ) + .bind(item.community_id.as_uuid()) + .bind(&item.event_id) + .bind(&item.claimed_by) + .execute(pool) + .await?; + Ok(result.rows_affected() == 1) +} + +/// Release a failed handoff with bounded retry delay. +pub async fn release_workflow_event_outbox( + pool: &PgPool, + item: &WorkflowEventOutboxItem, +) -> Result { + let result = sqlx::query( + "UPDATE workflow_event_outbox SET claimed_by = NULL, claimed_until = NULL, available_at = NOW() + INTERVAL '5 seconds' \ + WHERE community_id = $1 AND event_id = $2 AND claimed_by = $3", + ) + .bind(item.community_id.as_uuid()) + .bind(&item.event_id) + .bind(&item.claimed_by) + .execute(pool) + .await?; + Ok(result.rows_affected() == 1) +} + // -- Approval CRUD ------------------------------------------------------------ /// Parameters for creating a new approval request. @@ -1169,6 +1840,12 @@ fn row_to_run_record(row: sqlx::postgres::PgRow) -> Result { completed_at: row.try_get("completed_at")?, error_message: row.try_get("error_message")?, created_at: row.try_get("created_at")?, + lease_owner: row.try_get("lease_owner")?, + lease_token: row.try_get("lease_token")?, + lease_expires_at: row.try_get("lease_expires_at")?, + attempt: row.try_get("attempt")?, + available_at: row.try_get("available_at")?, + recovery_classification: row.try_get("recovery_classification")?, }) } @@ -1473,6 +2150,12 @@ mod tests { completed_at: None, error_message: None, created_at: now, + lease_owner: None, + lease_token: None, + lease_expires_at: None, + attempt: 0, + available_at: now, + recovery_classification: None, }; assert_eq!(record.id, id); @@ -1501,6 +2184,12 @@ mod tests { completed_at: None, error_message: None, created_at: now, + lease_owner: None, + lease_token: None, + lease_expires_at: None, + attempt: 0, + available_at: now, + recovery_classification: None, }; assert!(record.trigger_event_id.is_none()); @@ -1524,6 +2213,12 @@ mod tests { completed_at: Some(now), error_message: Some("step timeout exceeded".to_owned()), created_at: now, + lease_owner: None, + lease_token: None, + lease_expires_at: None, + attempt: 0, + available_at: now, + recovery_classification: Some("non_retryable".to_owned()), }; assert_eq!(record.status, RunStatus::Failed); @@ -1555,6 +2250,12 @@ mod tests { completed_at: Some(now), error_message: None, created_at: now, + lease_owner: None, + lease_token: None, + lease_expires_at: None, + attempt: 0, + available_at: now, + recovery_classification: None, }; assert!(record.execution_trace.is_array()); @@ -1577,6 +2278,12 @@ mod tests { completed_at: None, error_message: None, created_at: now, + lease_owner: None, + lease_token: None, + lease_expires_at: None, + attempt: 0, + available_at: now, + recovery_classification: None, }; let mut cloned = record.clone(); @@ -1891,6 +2598,321 @@ mod tests { ); } + /// A live lease is exclusive, and an expired worker cannot fence in a + /// replacement worker or finalize its run. + #[tokio::test] + #[ignore = "requires Postgres"] + async fn lease_fencing_rejects_stale_worker() { + let pool = setup_pool().await; + let community = make_community(&pool).await; + let (workflow_id, _) = make_workflow_in(&pool, community).await; + let trigger_id = Uuid::new_v4().as_bytes().to_vec(); + let run_id = create_workflow_run(&pool, community, workflow_id, Some(&trigger_id), None) + .await + .expect("create run"); + + let first = claim_workflow_run(&pool, community, "worker-a", 1) + .await + .expect("first claim query") + .expect("first worker claims run"); + assert_eq!(first.run_id, run_id); + assert!(claim_workflow_run(&pool, community, "worker-b", 1) + .await + .expect("second claim query") + .is_none()); + + tokio::time::sleep(std::time::Duration::from_millis(1_100)).await; + assert!(!heartbeat_workflow_run(&pool, &first, 30) + .await + .expect("stale heartbeat query")); + + let second = claim_workflow_run(&pool, community, "worker-b", 30) + .await + .expect("reclaim query") + .expect("expired run is reclaimable"); + assert_ne!(first.token, second.token); + assert!(!finalize_workflow_run( + &pool, + &first, + RunStatus::Completed, + 1, + &serde_json::json!([]), + None, + None, + ) + .await + .expect("stale finalize query")); + assert!(finalize_workflow_run( + &pool, + &second, + RunStatus::Completed, + 1, + &serde_json::json!([]), + None, + None, + ) + .await + .expect("current finalize query")); + assert!(claim_workflow_run(&pool, community, "worker-c", 30) + .await + .expect("terminal claim query") + .is_none()); + } + + /// Step attempts and action reservations survive retries and never produce + /// a second outbound identity for the same idempotency key. + #[tokio::test] + #[ignore = "requires Postgres"] + async fn step_attempts_and_action_effects_are_idempotent() { + let pool = setup_pool().await; + let community = make_community(&pool).await; + let (workflow_id, _) = make_workflow_in(&pool, community).await; + let trigger_id = Uuid::new_v4().as_bytes().to_vec(); + let run_id = create_workflow_run(&pool, community, workflow_id, Some(&trigger_id), None) + .await + .expect("create run"); + + let first = start_workflow_step_attempt(&pool, community, run_id, 0, "send") + .await + .expect("start first step attempt"); + assert_eq!(first.attempt, 1); + let recovered = start_workflow_step_attempt(&pool, community, run_id, 0, "send") + .await + .expect("recover in-flight step attempt"); + assert_eq!(recovered, first, "recovery must reuse the in-flight key"); + assert!(complete_workflow_step_attempt( + &pool, + community, + run_id, + 0, + first.attempt, + "completed", + Some(&serde_json::json!({"event": "published"})), + None, + None, + ) + .await + .expect("complete first step attempt")); + + let second = start_workflow_step_attempt(&pool, community, run_id, 0, "send") + .await + .expect("start retry step attempt"); + assert_eq!(second.attempt, 2); + assert_ne!(first.idempotency_key, second.idempotency_key); + + let key = "run-step-action-1"; + let reservation = reserve_workflow_action_effect(&pool, community, run_id, key) + .await + .expect("reserve action effect"); + assert!(reservation.newly_reserved); + assert!(reservation.event_id.is_none()); + let event_id = vec![0x42; 32]; + assert!( + complete_workflow_action_effect(&pool, community, key, &event_id) + .await + .expect("persist event identity") + ); + + let duplicate = reserve_workflow_action_effect(&pool, community, run_id, key) + .await + .expect("reuse action effect"); + assert!(!duplicate.newly_reserved); + assert_eq!(duplicate.event_id, Some(event_id)); + assert!(!complete_workflow_step_attempt( + &pool, + community, + run_id, + 0, + first.attempt, + "failed", + None, + Some("late duplicate completion"), + Some("ambiguous"), + ) + .await + .expect("duplicate completion query")); + } + + /// A delay releases the lease and is represented by `available_at`, so a + /// worker restart cannot lose it or keep a detached task asleep in memory. + #[tokio::test] + #[ignore = "requires Postgres"] + async fn deferred_run_is_not_claimable_until_available() { + let pool = setup_pool().await; + let community = make_community(&pool).await; + let (workflow_id, _) = make_workflow_in(&pool, community).await; + let trigger_id = Uuid::new_v4().as_bytes().to_vec(); + let run_id = create_workflow_run(&pool, community, workflow_id, Some(&trigger_id), None) + .await + .expect("create run"); + let lease = claim_workflow_run(&pool, community, "delay-worker", 30) + .await + .expect("claim query") + .expect("claim run"); + let available_at = Utc::now() + chrono::Duration::seconds(30); + assert!(defer_workflow_run( + &pool, + &lease, + 1, + &serde_json::json!([{"status": "deferred"}]), + available_at, + ) + .await + .expect("defer query")); + + let run = get_workflow_run(&pool, community, run_id) + .await + .expect("read deferred run"); + assert_eq!(run.status, RunStatus::Pending); + assert_eq!(run.current_step, 1); + assert!(run.available_at >= available_at - chrono::Duration::seconds(1)); + assert!(claim_workflow_run(&pool, community, "restart-worker", 30) + .await + .expect("early reclaim query") + .is_none()); + } + + /// Event-trigger handoffs are durable: a failed consumer can release its + /// claim and a replacement consumer can reclaim it, while only the owner + /// may acknowledge the handoff. + #[tokio::test] + #[ignore = "requires Postgres"] + async fn event_outbox_claim_release_and_ack_are_fenced() { + let pool = setup_pool().await; + let community = make_community(&pool).await; + let event_id = Uuid::new_v4().as_bytes().to_vec(); + sqlx::query("INSERT INTO workflow_event_outbox (community_id, event_id) VALUES ($1, $2)") + .bind(community.as_uuid()) + .bind(&event_id) + .execute(&pool) + .await + .expect("insert outbox item"); + + let claimed = claim_workflow_event_outbox(&pool, "consumer-a", 10) + .await + .expect("claim outbox") + .pop() + .expect("one outbox item"); + assert!(!ack_workflow_event_outbox( + &pool, + &WorkflowEventOutboxItem { + claimed_by: "consumer-b".into(), + ..claimed.clone() + } + ) + .await + .expect("wrong-owner ack query")); + assert!(release_workflow_event_outbox(&pool, &claimed) + .await + .expect("release outbox")); + sqlx::query( + "UPDATE workflow_event_outbox SET available_at = NOW() \ + WHERE community_id = $1 AND event_id = $2", + ) + .bind(community.as_uuid()) + .bind(&event_id) + .execute(&pool) + .await + .expect("make released item immediately claimable"); + let reclaimed = claim_workflow_event_outbox(&pool, "consumer-b", 10) + .await + .expect("reclaim outbox") + .pop() + .expect("released item is reclaimable"); + assert!(ack_workflow_event_outbox(&pool, &reclaimed) + .await + .expect("owner ack outbox")); + } + + /// Background reconciliation must not claim an approval-gated run; only + /// the explicit approval-resume path may transition it back to running. + #[tokio::test] + #[ignore = "requires Postgres"] + async fn scheduler_cannot_bypass_waiting_approval() { + let pool = setup_pool().await; + let community = make_community(&pool).await; + let (workflow_id, _) = make_workflow_in(&pool, community).await; + let trigger_id = Uuid::new_v4().as_bytes().to_vec(); + let run_id = create_workflow_run(&pool, community, workflow_id, Some(&trigger_id), None) + .await + .expect("create run"); + sqlx::query("UPDATE workflow_runs SET status = 'waiting_approval' WHERE community_id = $1 AND id = $2") + .bind(community.as_uuid()) + .bind(run_id) + .execute(&pool) + .await + .expect("set waiting approval"); + + assert!(claim_workflow_run(&pool, community, "scheduler", 30) + .await + .expect("scheduler claim query") + .is_none()); + assert!( + claim_waiting_approval_workflow_run(&pool, community, run_id, "approver", 30) + .await + .expect("approval claim query") + .is_some() + ); + } + + /// The event outbox trigger is selective: ordinary channel traffic must + /// not create durable workflow handoffs when no active workflow exists. + #[tokio::test] + #[ignore = "requires Postgres"] + async fn event_outbox_trigger_only_enqueues_workflow_channels() { + let pool = setup_pool().await; + let community = make_community(&pool).await; + let owner = vec![0xc3; 32]; + ensure_user(&pool, community, &owner) + .await + .expect("ensure owner"); + let ordinary_channel = make_channel(&pool, community, &owner).await; + let (workflow_id, workflow_community) = make_workflow_in(&pool, community).await; + let workflow = get_workflow(&pool, workflow_community, workflow_id) + .await + .expect("get workflow"); + let workflow_channel = workflow.channel_id.expect("workflow channel"); + + async fn insert_channel_event(pool: &PgPool, community: CommunityId, channel: Uuid) { + sqlx::query( + "INSERT INTO events (community_id, id, pubkey, created_at, kind, tags, content, sig, channel_id) \ + VALUES ($1, $2, $3, NOW(), 9, '[]', 'qa', $4, $5)", + ) + .bind(community.as_uuid()) + .bind(Uuid::new_v4().as_bytes().to_vec()) + .bind(vec![0xd4u8; 32]) + .bind(vec![0xe5u8; 64]) + .bind(channel) + .execute(pool) + .await + .expect("insert event"); + } + + insert_channel_event(&pool, community, ordinary_channel).await; + assert_eq!( + sqlx::query_scalar::<_, i64>( + "SELECT count(*) FROM workflow_event_outbox WHERE community_id = $1", + ) + .bind(community.as_uuid()) + .fetch_one(&pool) + .await + .expect("count ordinary outbox"), + 0 + ); + + insert_channel_event(&pool, community, workflow_channel).await; + assert_eq!( + sqlx::query_scalar::<_, i64>( + "SELECT count(*) FROM workflow_event_outbox WHERE community_id = $1", + ) + .bind(community.as_uuid()) + .fetch_one(&pool) + .await + .expect("count workflow outbox"), + 1 + ); + } + /// `attach_scheduled_workflow_run` links a won claim to the run it created. /// This is the regression for the missing `scheduled_workflow_fires. /// workflow_run_id` column: before the schema added it, the UPDATE failed at diff --git a/crates/buzz-relay/src/handlers/command_executor.rs b/crates/buzz-relay/src/handlers/command_executor.rs index 2d82736807..2444bd93df 100644 --- a/crates/buzz-relay/src/handlers/command_executor.rs +++ b/crates/buzz-relay/src/handlers/command_executor.rs @@ -889,7 +889,7 @@ async fn handle_workflow_trigger( // Persist the command event under the workflow channel even though the // trigger event itself only carries the workflow UUID. Storing channel // triggers as global events leaks workflow IDs to unrelated relay members. - let tx = match persist_command_event(state, tenant, event, workflow.channel_id).await? { + let mut tx = match persist_command_event(state, tenant, event, workflow.channel_id).await? { PersistResult::Duplicate => { return Ok(IngestResult { event_id: event.id.to_hex(), @@ -923,16 +923,15 @@ async fn handle_workflow_trigger( let trigger_ctx_json = serde_json::to_value(&trigger_ctx).ok(); 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(), - ) - .await - .map_err(|e| IngestError::Internal(format!("error: db create_workflow_run: {e}")))?; + let run_id = buzz_db::workflow::create_workflow_run_on_connection( + tx.as_mut(), + community_id, + workflow_id, + Some(&event_id_bytes), + trigger_ctx_json.as_ref(), + ) + .await + .map_err(|e| IngestError::Internal(format!("error: db create_workflow_run: {e}")))?; // Commit: event + run creation succeeded atomically. tx.commit() diff --git a/crates/buzz-relay/src/workflow_sink.rs b/crates/buzz-relay/src/workflow_sink.rs index 97c31c2561..59ccdc2b01 100644 --- a/crates/buzz-relay/src/workflow_sink.rs +++ b/crates/buzz-relay/src/workflow_sink.rs @@ -173,13 +173,16 @@ impl ActionSink for RelayActionSink { fn send_message( &self, community_id: CommunityId, + run_id: Uuid, channel_id: &str, text: &str, author_pubkey: &str, + idempotency_key: &str, ) -> Pin> + Send + '_>> { let channel_id = channel_id.to_owned(); let text = text.to_owned(); let author_pubkey = author_pubkey.to_owned(); + let idempotency_key = idempotency_key.to_owned(); Box::pin(async move { // 0. Upgrade weak reference — fails only during shutdown. @@ -188,6 +191,20 @@ impl ActionSink for RelayActionSink { .upgrade() .ok_or_else(|| ActionSinkError::Database("relay is shutting down".into()))?; + let reserved = state + .db + .reserve_workflow_action_effect(community_id, run_id, &idempotency_key) + .await + .map_err(|e| ActionSinkError::Database(e.to_string()))?; + if let Some(event_id) = reserved.event_id { + return Ok(hex::encode(event_id)); + } + if !reserved.newly_reserved { + return Err(ActionSinkError::Database( + "workflow action effect is reserved without a published event; recovery required".into(), + )); + } + // The run carries its owning community (`community_id`); the // relay-signed kind:9 message belongs to *that* community, never the // deployment default. Re-deriving the tenant from `config.relay_url` @@ -336,8 +353,10 @@ impl ActionSink for RelayActionSink { let (stored_event, was_inserted) = state .db - .insert_event_with_thread_metadata( + .insert_workflow_action_event( tenant.community(), + run_id, + &idempotency_key, &event, Some(channel_uuid), thread_meta, @@ -670,12 +689,37 @@ mod integration_tests { .expect("add agent member"); let sink = RelayActionSink::new(&state); + let workflow_id = state + .db + .create_workflow( + community, + Some(channel.id), + &author.public_key().to_bytes(), + "workflow-action-test", + r#"{"trigger":{"on":"schedule","cron":"0 0 * * *"},"steps":[]}"#, + &[0u8; 32], + ) + .await + .expect("create workflow"); + let run_id = state + .db + .create_workflow_run(community, workflow_id, None, None) + .await + .expect("create workflow run"); + let idempotency_key = "workflow-test-action"; + state + .db + .reserve_workflow_action_effect(community, run_id, idempotency_key) + .await + .expect("reserve workflow action effect"); let event_id_hex = sink .send_message( community, + run_id, &channel.id.to_string(), "heads up @Robby — please take a look", &author_hex, + idempotency_key, ) .await .expect("send_message"); diff --git a/crates/buzz-workflow/Cargo.toml b/crates/buzz-workflow/Cargo.toml index d4813e56d4..10d4b0f273 100644 --- a/crates/buzz-workflow/Cargo.toml +++ b/crates/buzz-workflow/Cargo.toml @@ -23,6 +23,7 @@ uuid = { workspace = true } chrono = { workspace = true } tokio = { workspace = true } tracing = { workspace = true } +metrics = { workspace = true } thiserror = { workspace = true } reqwest = { workspace = true, optional = true } diff --git a/crates/buzz-workflow/src/action_sink.rs b/crates/buzz-workflow/src/action_sink.rs index 0c6002e74e..535d3104c0 100644 --- a/crates/buzz-workflow/src/action_sink.rs +++ b/crates/buzz-workflow/src/action_sink.rs @@ -62,8 +62,10 @@ pub trait ActionSink: Send + Sync { fn send_message( &self, community_id: CommunityId, + run_id: uuid::Uuid, channel_id: &str, text: &str, author_pubkey: &str, + idempotency_key: &str, ) -> Pin> + Send + '_>>; } diff --git a/crates/buzz-workflow/src/executor.rs b/crates/buzz-workflow/src/executor.rs index e30541377e..a6c137bbf6 100644 --- a/crates/buzz-workflow/src/executor.rs +++ b/crates/buzz-workflow/src/executor.rs @@ -461,6 +461,11 @@ pub enum StepResult { /// Token used to resume or reject this approval gate. approval_token: String, }, + /// Step completed its durable scheduling boundary and should resume later. + Deferred { + /// Earliest time the next step may be claimed. + available_at: chrono::DateTime, + }, /// Step was skipped due to `if:` condition being false. Skipped, } @@ -518,6 +523,7 @@ fn resolve_send_message_channel( /// persist state and stop the execution loop. pub async fn dispatch_action( step_id: &str, + idempotency_key: &str, action: &ActionDef, engine: &WorkflowEngine, community_id: CommunityId, @@ -567,7 +573,14 @@ pub async fn dispatch_action( let event_id = engine .action_sink()? - .send_message(community_id, &channel_id, text, &owner_pubkey_hex) + .send_message( + community_id, + run_id, + &channel_id, + text, + &owner_pubkey_hex, + idempotency_key, + ) .await .map_err(WorkflowError::from)?; @@ -670,21 +683,10 @@ pub async fn dispatch_action( Delay { duration } => { let secs = parse_duration_secs(duration)?; - // Cap delay at 270 seconds (4.5 minutes) — must be less than default_timeout_secs (300s) - // to avoid non-deterministic StepTimeout. Long delays (hours/days) - // should use the scheduled resume pattern (future work: WF-09). - const MAX_DELAY_SECS: u64 = 270; - if secs > MAX_DELAY_SECS { - return Err(WorkflowError::InvalidDefinition(format!( - "delay exceeds maximum of {MAX_DELAY_SECS} seconds (got {secs}s); \ - use the scheduled resume pattern for long delays" - ))); - } info!(run_id = %run_id, step = step_id, "Delay {duration} ({secs}s)"); - tokio::time::sleep(std::time::Duration::from_secs(secs)).await; - Ok(StepResult::Completed( - serde_json::json!({ "slept_secs": secs }), - )) + Ok(StepResult::Deferred { + available_at: chrono::Utc::now() + chrono::Duration::seconds(secs as i64), + }) } } } @@ -949,6 +951,8 @@ pub struct ExecutionResult { pub step_outputs: HashMap, /// Execution trace: one entry per completed/skipped step. pub trace: Vec, + /// Set when the run has been durably deferred until a future timestamp. + pub reschedule_at: Option>, } /// Execute a workflow run sequentially. @@ -974,7 +978,40 @@ pub async fn execute_run( def: &WorkflowDef, trigger_ctx: &TriggerContext, ) -> Result { - // Fail fast if all concurrency permits are in use — no queuing. + let lease = engine + .db + .claim_specific_workflow_run(community_id, run_id, &engine.worker_id, 300) + .await + .map_err(|e| { + ( + WorkflowError::from(e), + crate::error::PartialProgress::default(), + ) + })? + .ok_or_else(|| { + ( + WorkflowError::InvalidDefinition( + "workflow run is not claimable or is owned by another worker".into(), + ), + crate::error::PartialProgress::default(), + ) + })?; + if lease.run_id != run_id { + return Err(( + WorkflowError::InvalidDefinition( + "claimed workflow run did not match requested run".into(), + ), + crate::error::PartialProgress::default(), + )); + } + engine + .active_leases + .insert((community_id, run_id), lease.clone()); + metrics::counter!("buzz_workflow_run_claims_total").increment(1); + engine.start_lease_heartbeat(lease); + + // Claim before checking capacity so a capacity rejection still owns a + // fencing lease and can be durably rescheduled by finalize_run. let _permit = engine.run_semaphore.try_acquire().map_err(|_| { ( WorkflowError::CapacityExceeded, @@ -982,16 +1019,9 @@ pub async fn execute_run( ) })?; - engine + let run = engine .db - .update_workflow_run( - community_id, - run_id, - buzz_db::workflow::RunStatus::Running, - 0, - &serde_json::json!([]), - None, - ) + .get_workflow_run(community_id, run_id) .await .map_err(|e| { ( @@ -999,8 +1029,20 @@ pub async fn execute_run( crate::error::PartialProgress::default(), ) })?; - - execute_steps(engine, community_id, run_id, def, trigger_ctx, 0, None).await + let start_step = run.current_step.max(0) as usize; + let initial_trace = run.execution_trace.as_array().cloned().unwrap_or_default(); + let initial_outputs = outputs_from_trace(&initial_trace); + execute_steps( + engine, + community_id, + run_id, + def, + trigger_ctx, + start_step, + Some(initial_outputs), + Some(initial_trace), + ) + .await } /// Resume execution from a specific step index (used for approval resume). @@ -1024,43 +1066,47 @@ pub async fn execute_from_step( start_index: usize, initial_outputs: Option>, ) -> Result { - // Fail fast if all concurrency permits are in use — no queuing. - let _permit = engine.run_semaphore.try_acquire().map_err(|_| { - ( - WorkflowError::CapacityExceeded, - crate::error::PartialProgress::default(), - ) - })?; - - // Mark run as Running now that we have a permit (resume from approval). - // Preserve the existing execution trace from pre-approval steps. - let existing_trace = match engine.db.get_workflow_run(community_id, run_id).await { - Ok(r) => r.execution_trace, - Err(e) => { - warn!( - run_id = %run_id, - "Failed to read existing trace for resume — pre-approval trace will be lost: {e}" - ); - serde_json::json!([]) - } - }; - engine + // Claim the waiting run with a fresh fencing token only after approval. + let lease = engine .db - .update_workflow_run( - community_id, - run_id, - buzz_db::workflow::RunStatus::Running, - start_index as i32, - &existing_trace, - None, - ) + .claim_waiting_approval_workflow_run(community_id, run_id, &engine.worker_id, 300) .await .map_err(|e| { ( WorkflowError::from(e), crate::error::PartialProgress::default(), ) + })? + .ok_or_else(|| { + ( + WorkflowError::InvalidDefinition( + "workflow run is not claimable or is owned by another worker".into(), + ), + crate::error::PartialProgress::default(), + ) })?; + if lease.run_id != run_id { + return Err(( + WorkflowError::InvalidDefinition( + "claimed workflow run did not match requested run".into(), + ), + crate::error::PartialProgress::default(), + )); + } + engine + .active_leases + .insert((community_id, run_id), lease.clone()); + metrics::counter!("buzz_workflow_run_claims_total").increment(1); + engine.start_lease_heartbeat(lease); + + // As in execute_run, keep capacity failures lease-protected so they can + // be persisted as retryable rather than stranded in `running`. + let _permit = engine.run_semaphore.try_acquire().map_err(|_| { + ( + WorkflowError::CapacityExceeded, + crate::error::PartialProgress::default(), + ) + })?; execute_steps( engine, @@ -1070,16 +1116,30 @@ pub async fn execute_from_step( trigger_ctx, start_index, initial_outputs, + None, ) .await } +fn outputs_from_trace(trace: &[JsonValue]) -> HashMap { + trace + .iter() + .filter_map(|entry| { + Some(( + entry.get("step_id")?.as_str()?.to_owned(), + entry.get("output")?.clone(), + )) + }) + .collect() +} + /// Internal: execute workflow steps starting from `start_index`, without /// acquiring the semaphore. Called by both [`execute_run`] and /// [`execute_from_step`] after they have already acquired a permit. /// /// On error, returns `(WorkflowError, PartialProgress)` so callers can persist /// the trace of steps completed before the failure. +#[allow(clippy::too_many_arguments)] async fn execute_steps( engine: &WorkflowEngine, community_id: CommunityId, @@ -1088,9 +1148,10 @@ async fn execute_steps( trigger_ctx: &TriggerContext, start_index: usize, initial_outputs: Option>, + initial_trace: Option>, ) -> Result { let mut step_outputs: HashMap = initial_outputs.unwrap_or_default(); - let mut trace: Vec = Vec::new(); + let mut trace: Vec = initial_trace.unwrap_or_default(); for (i, step) in def.steps.iter().enumerate() { if i < start_index { @@ -1136,10 +1197,24 @@ async fn execute_steps( let timeout_secs = step .timeout_secs .unwrap_or(engine.config.default_timeout_secs); + let attempt = engine + .db + .start_workflow_step_attempt(community_id, run_id, i as i32, &step.id) + .await + .map_err(|e| { + ( + WorkflowError::from(e), + crate::error::PartialProgress { + step_index: i, + trace: trace.clone(), + }, + ) + })?; let dispatch_result = tokio::time::timeout( std::time::Duration::from_secs(timeout_secs), dispatch_action( &step.id, + &attempt.idempotency_key, &resolved_action, engine, community_id, @@ -1152,6 +1227,24 @@ async fn execute_steps( let result = match dispatch_result { Ok(Ok(r)) => r, Ok(Err(e)) => { + let recovery = if e.to_string().contains("recovery required") { + "ambiguous" + } else { + "unknown" + }; + let _ = engine + .db + .complete_workflow_step_attempt( + community_id, + run_id, + i as i32, + attempt.attempt, + "failed", + None, + Some(&e.to_string()), + Some(recovery), + ) + .await; let progress = crate::error::PartialProgress { step_index: i, trace, @@ -1159,6 +1252,20 @@ async fn execute_steps( return Err((e, progress)); } Err(_timeout) => { + let timeout_error = format!("step {} timed out after {timeout_secs}s", step.id); + let _ = engine + .db + .complete_workflow_step_attempt( + community_id, + run_id, + i as i32, + attempt.attempt, + "failed", + None, + Some(&timeout_error), + Some("unknown"), + ) + .await; let progress = crate::error::PartialProgress { step_index: i, trace, @@ -1175,6 +1282,19 @@ async fn execute_steps( match result { StepResult::Completed(output) => { + let _ = engine + .db + .complete_workflow_step_attempt( + community_id, + run_id, + i as i32, + attempt.attempt, + "completed", + Some(&output), + None, + None, + ) + .await; debug!(run_id = %run_id, step = %step.id, "Step completed"); trace.push(serde_json::json!({ "step_id": step.id, @@ -1184,6 +1304,19 @@ async fn execute_steps( step_outputs.insert(step.id.clone(), output); } StepResult::Suspended { approval_token } => { + let _ = engine + .db + .complete_workflow_step_attempt( + community_id, + run_id, + i as i32, + attempt.attempt, + "suspended", + None, + None, + Some("ambiguous"), + ) + .await; info!( run_id = %run_id, step = %step.id, "Step suspended — awaiting approval (token: )" @@ -1195,9 +1328,50 @@ async fn execute_steps( step_index: i, step_outputs, trace, + reschedule_at: None, + }); + } + StepResult::Deferred { available_at } => { + let _ = engine + .db + .complete_workflow_step_attempt( + community_id, + run_id, + i as i32, + attempt.attempt, + "deferred", + Some(&serde_json::json!({ "available_at": available_at })), + None, + Some("delayed"), + ) + .await; + trace.push(serde_json::json!({ + "step_id": step.id, + "status": "deferred", + "available_at": available_at, + })); + return Ok(ExecutionResult { + approval_token: None, + step_index: i + 1, + step_outputs, + trace, + reschedule_at: Some(available_at), }); } StepResult::Skipped => { + let _ = engine + .db + .complete_workflow_step_attempt( + community_id, + run_id, + i as i32, + attempt.attempt, + "skipped", + None, + None, + None, + ) + .await; debug!(run_id = %run_id, step = %step.id, "Step skipped"); trace.push(serde_json::json!({ "step_id": step.id, @@ -1213,6 +1387,7 @@ async fn execute_steps( step_index: def.steps.len(), step_outputs, trace, + reschedule_at: None, }) } @@ -1801,6 +1976,17 @@ mod tests { assert!(ctx.webhook_fields.is_empty()); } + #[test] + fn recovery_trace_reconstructs_step_outputs() { + let trace = vec![ + json!({"step_id": "fetch", "status": "completed", "output": {"value": 42}}), + json!({"step_id": "send", "status": "deferred"}), + ]; + let outputs = outputs_from_trace(&trace); + assert_eq!(outputs.get("fetch"), Some(&json!({"value": 42}))); + assert!(!outputs.contains_key("send")); + } + #[test] fn send_message_uses_bound_workflow_channel_by_default() { let workflow_channel_id = Uuid::new_v4(); diff --git a/crates/buzz-workflow/src/lib.rs b/crates/buzz-workflow/src/lib.rs index 7aaa3d1702..204a9c8299 100644 --- a/crates/buzz-workflow/src/lib.rs +++ b/crates/buzz-workflow/src/lib.rs @@ -33,6 +33,7 @@ pub mod action_sink; pub mod error; pub mod executor; +pub mod recovery; pub mod schema; pub use action_sink::{ActionSink, ActionSinkError}; @@ -50,7 +51,7 @@ use buzz_db::workflow::RunStatus; use buzz_db::Db; use chrono::{DateTime, Utc}; use dashmap::DashMap; -use tokio::sync::Semaphore; +use tokio::sync::{Notify, Semaphore}; use uuid::Uuid; /// Runtime configuration for the workflow engine. @@ -102,6 +103,13 @@ pub struct WorkflowEngine { /// `buzz-relay`). pub(crate) workflow_cache: moka::sync::Cache<(CommunityId, Uuid), Arc>>, + /// Stable process identity used for lease ownership and fencing logs. + pub(crate) worker_id: String, + /// Leases claimed by this process, keyed by tenant and run. + pub(crate) active_leases: + Arc>, + /// Shutdown signal that stops new scheduler claims while existing tasks drain. + pub(crate) shutdown: Arc, } impl WorkflowEngine { @@ -119,6 +127,9 @@ impl WorkflowEngine { .max_capacity(10_000) .time_to_live(std::time::Duration::from_secs(10)) .build(), + worker_id: format!("buzz-workflow-{}", Uuid::new_v4()), + active_leases: Arc::new(DashMap::new()), + shutdown: Arc::new(Notify::new()), } } @@ -194,6 +205,144 @@ impl WorkflowEngine { }) } + #[allow(clippy::too_many_arguments)] + async fn persist_owned_run( + &self, + community_id: CommunityId, + run_id: Uuid, + status: RunStatus, + step: i32, + trace: &serde_json::Value, + error: Option<&str>, + recovery: Option<&str>, + ) { + if let Some((_, lease)) = self.active_leases.remove(&(community_id, run_id)) { + match self + .db + .finalize_workflow_run(&lease, status.clone(), step, trace, error, recovery) + .await + { + Ok(true) => { + metrics::counter!("buzz_workflow_lease_finalizations_total", "status" => status.to_string()).increment(1); + return; + } + Ok(false) => { + metrics::counter!("buzz_workflow_stale_finalize_rejections_total").increment(1); + tracing::warn!(run_id = %run_id, "stale workflow lease rejected finalization") + } + Err(e) => { + metrics::counter!("buzz_workflow_lease_finalize_errors_total").increment(1); + tracing::error!(run_id = %run_id, "lease-protected workflow finalization failed: {e}") + } + } + return; + } + tracing::warn!( + run_id = %run_id, + "workflow finalization skipped because the fencing lease is absent" + ); + } + + async fn defer_owned_run( + &self, + community_id: CommunityId, + run_id: Uuid, + step: i32, + trace: &serde_json::Value, + available_at: DateTime, + ) { + if let Some((_, lease)) = self.active_leases.remove(&(community_id, run_id)) { + match self + .db + .defer_workflow_run(&lease, step, trace, available_at) + .await + { + Ok(true) => { + metrics::counter!("buzz_workflow_run_deferrals_total").increment(1); + } + Ok(false) => { + metrics::counter!("buzz_workflow_stale_defer_rejections_total").increment(1); + tracing::warn!(run_id = %run_id, "stale workflow lease rejected delay deferral"); + } + Err(e) => { + metrics::counter!("buzz_workflow_defer_errors_total").increment(1); + tracing::error!(run_id = %run_id, "durable workflow delay persistence failed: {e}"); + } + } + } else { + tracing::warn!(run_id = %run_id, "workflow delay completion had no active lease"); + } + } + + /// Keep a claimed run alive while its executor is making progress. The + /// loop exits as soon as finalization removes the lease from the active set. + pub(crate) fn start_lease_heartbeat(&self, lease: buzz_db::workflow::WorkflowRunLease) { + let db = self.db.clone(); + let active = Arc::clone(&self.active_leases); + tokio::spawn(async move { + let key = (lease.community_id, lease.run_id); + loop { + tokio::time::sleep(std::time::Duration::from_secs(60)).await; + let Some(current) = active.get(&key).map(|entry| entry.clone()) else { + break; + }; + if !db + .heartbeat_workflow_run(¤t, 300) + .await + .unwrap_or(false) + { + break; + } + } + }); + } + + /// Stop scheduling new work while leaving active runs reclaimable. + pub fn request_shutdown(&self) { + // `notify_one` retains a permit when the scheduler is between ticks, + // so a shutdown request cannot be lost while the run loop is doing DB + // work rather than currently waiting in `select!`. + self.shutdown.notify_one(); + } + + /// Drain durable event-trigger handoffs. The normal post-store hook still + /// provides low latency; this path closes the crash window between event + /// commit and handler execution. + async fn drain_event_outbox(self: &Arc) { + let items = match self + .db + .claim_workflow_event_outbox(&self.worker_id, 100) + .await + { + Ok(items) => items, + Err(e) => { + tracing::debug!("workflow event outbox unavailable: {e}"); + return; + } + }; + for item in items { + let processed = match self + .db + .get_event_by_id(item.community_id, &item.event_id) + .await + { + Ok(Some(event)) => self.on_event(item.community_id, &event).await.is_ok(), + Ok(None) => true, + Err(e) => { + tracing::warn!(event_id = ?item.event_id, "workflow outbox event read failed: {e}"); + false + } + }; + if processed { + metrics::counter!("buzz_workflow_event_outbox_acked_total").increment(1); + let _ = self.db.ack_workflow_event_outbox(&item).await; + } else { + metrics::counter!("buzz_workflow_event_outbox_retries_total").increment(1); + let _ = self.db.release_workflow_event_outbox(&item).await; + } + } + } + /// Parse and validate a YAML workflow definition. /// /// Returns `(WorkflowDef, canonical_json)` on success. The canonical JSON @@ -226,7 +375,16 @@ impl WorkflowEngine { let trace_json = serde_json::Value::Array(full_trace); let step_count = result.step_index as i32; - if result.approval_token.is_some() { + if let Some(available_at) = result.reschedule_at { + self.defer_owned_run( + community_id, + run_id, + step_count, + &trace_json, + available_at, + ) + .await; + } else if result.approval_token.is_some() { // Approval gates are not yet implemented (WF-08). // Fail explicitly rather than creating unreachable WaitingApproval rows. tracing::warn!( @@ -234,42 +392,28 @@ impl WorkflowEngine { step_index = result.step_index, "Workflow hit approval gate — not yet implemented, marking as failed" ); - if let Err(e) = self - .db - .update_workflow_run( - community_id, - run_id, - RunStatus::Failed, - step_count, - &trace_json, - Some("approval gates not yet implemented — see WF-08"), - ) - .await - { - tracing::error!( - run_id = %run_id, - "Failed to update run to Failed (approval gate): {e}" - ); - } + self.persist_owned_run( + community_id, + run_id, + RunStatus::Failed, + step_count, + &trace_json, + Some("approval gates not yet implemented — see WF-08"), + Some("non_retryable"), + ) + .await; } else { tracing::info!(run_id = %run_id, "Workflow run completed"); - if let Err(e) = self - .db - .update_workflow_run( - community_id, - run_id, - RunStatus::Completed, - step_count, - &trace_json, - None, - ) - .await - { - tracing::error!( - run_id = %run_id, - "Failed to update run to Completed: {e}" - ); - } + self.persist_owned_run( + community_id, + run_id, + RunStatus::Completed, + step_count, + &trace_json, + None, + None, + ) + .await; } } Err((e, progress)) => { @@ -277,23 +421,23 @@ impl WorkflowEngine { let mut full_trace = prefix; full_trace.extend(progress.trace); let trace_json = serde_json::Value::Array(full_trace); - if let Err(db_err) = self - .db - .update_workflow_run( - community_id, - run_id, - RunStatus::Failed, - progress.step_index as i32, - &trace_json, - Some(&e.to_string()), - ) - .await - { - tracing::error!( - run_id = %run_id, - "Failed to update run to Failed: {db_err}" - ); - } + let classification = if matches!(e, WorkflowError::CapacityExceeded) { + "retryable" + } else if e.to_string().contains("recovery required") { + "ambiguous" + } else { + "non_retryable" + }; + self.persist_owned_run( + community_id, + run_id, + RunStatus::Failed, + progress.step_index as i32, + &trace_json, + Some(&e.to_string()), + Some(classification), + ) + .await; } } } @@ -328,7 +472,13 @@ impl WorkflowEngine { let kind_u32 = event_kind_u32(&event.event); // Exclude workflow execution events to prevent infinite loops. - if is_workflow_execution_kind(kind_u32) { + if is_workflow_execution_kind(kind_u32) + || event.event.tags.iter().any(|tag| { + let parts = tag.as_slice(); + parts.first().map(String::as_str) == Some("buzz:workflow") + && parts.get(1).map(String::as_str) == Some("true") + }) + { return Ok(()); } @@ -408,8 +558,8 @@ impl WorkflowEngine { { Ok(id) => id, Err(e) => { - tracing::error!(workflow_id = %workflow.id, "Failed to create run: {e}"); - continue; + tracing::error!(workflow_id = %workflow.id, "Failed to create run; retaining event outbox item for retry: {e}"); + return Err(WorkflowError::from(e)); } }; @@ -484,10 +634,18 @@ impl WorkflowEngine { tracing::info!("WorkflowEngine cron loop started (60s tick)"); loop { - tokio::time::sleep(std::time::Duration::from_secs(60)).await; + tokio::select! { + _ = tokio::time::sleep(std::time::Duration::from_secs(60)) => {} + _ = self.shutdown.notified() => { + tracing::info!("WorkflowEngine scheduler stopping; active leases remain reclaimable"); + return; + } + } let now = Utc::now(); + self.drain_event_outbox().await; + let workflows = match self.db.list_all_enabled_workflows().await { Ok(wf) => wf, Err(e) => { @@ -615,33 +773,8 @@ impl WorkflowEngine { // input; the claim binds `(community_id, workflow_id, // scheduled_for)` so a duplicate workflow UUID in another // community claims independently. - match self - .db - .claim_scheduled_workflow_fire(community_id, workflow.id, scheduled_for) - .await - { - Ok(Some(_)) => {} - Ok(None) => { - // Another pod (or an earlier tick this pod) already - // claimed this instant. Still advance the in-memory - // interval clock so we don't re-attempt the claim every - // tick for the rest of the interval. - if trigger_type == "interval" { - self.last_fired.insert((community_id, workflow.id), now); - } - continue; - } - Err(e) => { - tracing::error!( - workflow_id = %workflow.id, - "Cron tick: scheduled-fire claim failed: {e}" - ); - continue; - } - } - - // Fix 5: handle serialization errors explicitly rather than silently - // dropping the trigger context with .ok(). + // Claim, create, and attach are one transaction. A failed run + // insert no longer strands a durable fire with no resumable run. let trigger_ctx = executor::TriggerContext { channel_id: channel_id.to_string(), timestamp: now.timestamp().to_string(), @@ -658,44 +791,36 @@ impl WorkflowEngine { } }; - let run_id = match self + let run_claim = match self .db - .create_workflow_run( + .claim_and_create_scheduled_workflow_run( community_id, workflow.id, - None, // no trigger event for cron + scheduled_for, trigger_ctx_json.as_ref(), ) .await { - Ok(id) => id, + Ok(Some(claim)) => claim, + Ok(None) => { + // Another pod (or an earlier tick this pod) already + // claimed this instant. Still advance the in-memory + // interval clock so we don't re-attempt the claim every + // tick for the rest of the interval. + if trigger_type == "interval" { + self.last_fired.insert((community_id, workflow.id), now); + } + continue; + } Err(e) => { tracing::error!( workflow_id = %workflow.id, - "Cron tick: failed to create workflow run: {e}" + "Cron tick: scheduled-fire claim failed: {e}" ); - // The claim is held but the run failed to create. The - // claim row intentionally stays (its `workflow_run_id` - // NULL) so this instant is not re-fired: at-most-once is - // preserved over exactly-once on transient run-insert - // failures. continue; } }; - - // Link the won claim to its run for ops/audit forensics. The - // claim row already guarantees dedupe; this is best-effort. - if let Err(e) = self - .db - .attach_scheduled_workflow_run(community_id, workflow.id, scheduled_for, run_id) - .await - { - tracing::warn!( - workflow_id = %workflow.id, - run_id = %run_id, - "Cron tick: failed to attach run to scheduled-fire claim: {e}" - ); - } + let run_id = run_claim.run_id; // Update last_fired AFTER a successful claim+insert so that a // failure doesn't suppress the next tick for the full interval. @@ -732,6 +857,58 @@ impl WorkflowEngine { }); } + // Restart reconciliation: detached tasks disappear with the + // process, but their pending/retryable/expired rows remain. Scan + // durable runs and re-submit them; execute_run performs the final + // fenced claim so duplicate scheduler pods are harmless. + for workflow in &workflows { + let def: schema::WorkflowDef = + match serde_json::from_value(workflow.definition.clone()) { + Ok(def) => def, + Err(_) => continue, + }; + let runs = match self + .db + .list_workflow_runs(workflow.community_id, workflow.id, 100) + .await + { + Ok(runs) => runs, + Err(e) => { + tracing::warn!(workflow_id = %workflow.id, "reconciliation run scan failed: {e}"); + continue; + } + }; + for run in runs { + let claimable = matches!(run.status, RunStatus::Pending) + || (matches!(run.status, RunStatus::Failed) + && run.recovery_classification.as_deref() == Some("retryable")) + || (matches!(run.status, RunStatus::Running) + && run + .lease_expires_at + .map(|expires| expires <= now) + .unwrap_or(false)); + if !claimable || run.available_at > now { + continue; + } + let ctx = run + .trigger_context + .as_ref() + .and_then(|value| serde_json::from_value(value.clone()).ok()) + .unwrap_or_default(); + let run_community = workflow.community_id; + let engine = Arc::clone(self); + let def_clone = def.clone(); + tokio::spawn(async move { + let result = + executor::execute_run(&engine, run_community, run.id, &def_clone, &ctx) + .await; + engine + .finalize_run(run_community, run.id, result, None) + .await; + }); + } + } + // Fix 1: prune stale last_fired entries for workflows that are no longer // active/enabled. Without this the DashMap grows monotonically as // workflows are deleted or disabled. Keyed by `(community_id, id)` so diff --git a/crates/buzz-workflow/src/recovery.rs b/crates/buzz-workflow/src/recovery.rs new file mode 100644 index 0000000000..a80ebcbae2 --- /dev/null +++ b/crates/buzz-workflow/src/recovery.rs @@ -0,0 +1,152 @@ +//! Crash-window classification for workflow recovery. +//! +//! The database lease and idempotency records are the source of truth. This +//! small, side-effect-free classifier documents the expected decision at each +//! failure boundary and keeps that policy executable in tests. + +/// The point at which a worker stopped making progress. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum CrashWindow { + /// The event/run was persisted, but no worker claimed it. + BeforeClaim, + /// A lease was claimed, but the step attempt was not started. + AfterClaimBeforeStep, + /// A step attempt was persisted and the worker stopped during execution. + DuringStep, + /// The remote side effect may have succeeded, but its durable receipt was + /// not written yet. + AfterEffectBeforeReceipt, + /// The run reached a terminal result, but finalization was interrupted. + BeforeFinalize, +} + +/// Durable facts available to the recovery decision. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub struct RecoveryFacts { + /// Whether the worker lease has expired (or the worker disappeared). + pub lease_expired: bool, + /// Whether a durable action-effect receipt exists. + pub effect_receipt_exists: bool, + /// Whether the run already has a terminal status. + pub run_is_terminal: bool, +} + +/// Recovery action selected by the worker after a crash/restart. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum RecoveryAction { + /// Claim or reclaim the run and continue from its durable step cursor. + ReclaimAndResume, + /// Resume the durable step attempt using its idempotency key. + ResumeIdempotentStep, + /// Do not repeat the remote effect; reconcile the existing receipt. + ReconcileEffect, + /// Leave the terminal result immutable. + PreserveTerminal, + /// Keep waiting for the current lease to expire. + WaitForLeaseExpiry, +} + +/// Classify a crash window from the records persisted by Buzz. +pub fn classify(window: CrashWindow, facts: RecoveryFacts) -> RecoveryAction { + if facts.run_is_terminal { + return RecoveryAction::PreserveTerminal; + } + if !facts.lease_expired { + return RecoveryAction::WaitForLeaseExpiry; + } + + match window { + CrashWindow::BeforeClaim | CrashWindow::AfterClaimBeforeStep => { + RecoveryAction::ReclaimAndResume + } + CrashWindow::DuringStep => RecoveryAction::ResumeIdempotentStep, + CrashWindow::AfterEffectBeforeReceipt => { + if facts.effect_receipt_exists { + RecoveryAction::ReconcileEffect + } else { + // The action sink must reserve by idempotency key before + // retrying; the sink decides whether the remote effect is + // safely replayable or requires operator reconciliation. + RecoveryAction::ResumeIdempotentStep + } + } + CrashWindow::BeforeFinalize => RecoveryAction::ReclaimAndResume, + } +} + +#[cfg(test)] +mod tests { + use super::*; + + const EXPIRED: RecoveryFacts = RecoveryFacts { + lease_expired: true, + effect_receipt_exists: false, + run_is_terminal: false, + }; + + #[test] + fn crash_windows_are_recoverable_after_lease_expiry() { + let cases = [ + (CrashWindow::BeforeClaim, RecoveryAction::ReclaimAndResume), + ( + CrashWindow::AfterClaimBeforeStep, + RecoveryAction::ReclaimAndResume, + ), + ( + CrashWindow::DuringStep, + RecoveryAction::ResumeIdempotentStep, + ), + ( + CrashWindow::AfterEffectBeforeReceipt, + RecoveryAction::ResumeIdempotentStep, + ), + ( + CrashWindow::BeforeFinalize, + RecoveryAction::ReclaimAndResume, + ), + ]; + for (window, expected) in cases { + assert_eq!(classify(window, EXPIRED), expected); + } + } + + #[test] + fn existing_effect_receipt_is_reconciled_without_replay() { + let facts = RecoveryFacts { + effect_receipt_exists: true, + ..EXPIRED + }; + assert_eq!( + classify(CrashWindow::AfterEffectBeforeReceipt, facts), + RecoveryAction::ReconcileEffect + ); + } + + #[test] + fn live_lease_is_never_stolen() { + assert_eq!( + classify( + CrashWindow::DuringStep, + RecoveryFacts { + lease_expired: false, + ..EXPIRED + } + ), + RecoveryAction::WaitForLeaseExpiry + ); + } + + #[test] + fn terminal_runs_are_immutable_even_if_replayed() { + assert_eq!( + classify( + CrashWindow::BeforeFinalize, + RecoveryFacts { + run_is_terminal: true, + ..EXPIRED + } + ), + RecoveryAction::PreserveTerminal + ); + } +} diff --git a/migrations/0026_workflow_run_leases.sql b/migrations/0026_workflow_run_leases.sql new file mode 100644 index 0000000000..6336050cd0 --- /dev/null +++ b/migrations/0026_workflow_run_leases.sql @@ -0,0 +1,24 @@ +-- Durable workflow execution ownership. A lease is fencing, not a liveness +-- hint: every worker mutation must present the current token. +ALTER TABLE workflow_runs + ADD COLUMN IF NOT EXISTS definition_snapshot JSONB, + ADD COLUMN IF NOT EXISTS definition_hash BYTEA, + ADD COLUMN IF NOT EXISTS lease_owner TEXT, + ADD COLUMN IF NOT EXISTS lease_token UUID, + ADD COLUMN IF NOT EXISTS lease_expires_at TIMESTAMPTZ, + ADD COLUMN IF NOT EXISTS heartbeat_at TIMESTAMPTZ, + ADD COLUMN IF NOT EXISTS attempt INTEGER NOT NULL DEFAULT 0, + ADD COLUMN IF NOT EXISTS available_at TIMESTAMPTZ NOT NULL DEFAULT NOW(), + ADD COLUMN IF NOT EXISTS recovery_classification TEXT; + +CREATE INDEX IF NOT EXISTS idx_workflow_runs_claimable + ON workflow_runs (community_id, available_at, created_at) + WHERE status IN ('pending', 'waiting_approval', 'failed', 'running'); + +CREATE INDEX IF NOT EXISTS idx_workflow_runs_lease_expiry + ON workflow_runs (lease_expires_at) + WHERE status = 'running'; + +CREATE UNIQUE INDEX IF NOT EXISTS uq_workflow_runs_trigger_identity + ON workflow_runs (community_id, workflow_id, trigger_event_id) + WHERE trigger_event_id IS NOT NULL; diff --git a/migrations/0027_workflow_step_attempts.sql b/migrations/0027_workflow_step_attempts.sql new file mode 100644 index 0000000000..65c99a0351 --- /dev/null +++ b/migrations/0027_workflow_step_attempts.sql @@ -0,0 +1,33 @@ +CREATE TABLE IF NOT EXISTS workflow_step_attempts ( + community_id UUID NOT NULL, + run_id UUID NOT NULL, + step_index INTEGER NOT NULL, + step_id VARCHAR(64) NOT NULL, + attempt INTEGER NOT NULL, + idempotency_key TEXT NOT NULL, + status TEXT NOT NULL, + started_at TIMESTAMPTZ NOT NULL DEFAULT NOW(), + completed_at TIMESTAMPTZ, + result JSONB, + error_message TEXT, + recovery_classification TEXT, + PRIMARY KEY (community_id, run_id, step_index, attempt), + UNIQUE (community_id, idempotency_key), + FOREIGN KEY (community_id, run_id) + REFERENCES workflow_runs (community_id, id) ON DELETE CASCADE +); + +CREATE INDEX IF NOT EXISTS idx_workflow_step_attempts_recovery + ON workflow_step_attempts (community_id, status, started_at); + +CREATE TABLE IF NOT EXISTS workflow_action_effects ( + community_id UUID NOT NULL, + run_id UUID NOT NULL, + idempotency_key TEXT NOT NULL, + event_id BYTEA, + status TEXT NOT NULL DEFAULT 'reserved', + created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(), + PRIMARY KEY (community_id, idempotency_key), + FOREIGN KEY (community_id, run_id) + REFERENCES workflow_runs (community_id, id) ON DELETE CASCADE +); diff --git a/migrations/0028_workflow_event_outbox.sql b/migrations/0028_workflow_event_outbox.sql new file mode 100644 index 0000000000..f240783395 --- /dev/null +++ b/migrations/0028_workflow_event_outbox.sql @@ -0,0 +1,41 @@ +CREATE TABLE IF NOT EXISTS workflow_event_outbox ( + community_id UUID NOT NULL, + event_id BYTEA NOT NULL, + available_at TIMESTAMPTZ NOT NULL DEFAULT NOW(), + attempts INTEGER NOT NULL DEFAULT 0, + claimed_by TEXT, + claimed_until TIMESTAMPTZ, + created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(), + PRIMARY KEY (community_id, event_id) +); + +CREATE INDEX IF NOT EXISTS idx_workflow_event_outbox_claimable + ON workflow_event_outbox (available_at, claimed_until); + +CREATE OR REPLACE FUNCTION enqueue_workflow_event_outbox() RETURNS trigger +LANGUAGE plpgsql AS $$ +BEGIN + -- Do not create a durable handoff for every ordinary channel message. + -- Only channels with an active workflow need event-trigger recovery; this + -- keeps the outbox proportional to workflow traffic rather than chat + -- traffic and avoids an unbounded backlog when no workflows are enabled. + IF NEW.channel_id IS NOT NULL AND EXISTS ( + SELECT 1 + FROM workflows w + WHERE w.community_id = NEW.community_id + AND w.channel_id = NEW.channel_id + AND w.status = 'active' + AND w.enabled + ) THEN + INSERT INTO workflow_event_outbox (community_id, event_id) + VALUES (NEW.community_id, NEW.id) + ON CONFLICT (community_id, event_id) DO NOTHING; + END IF; + RETURN NEW; +END +$$; + +DROP TRIGGER IF EXISTS events_workflow_outbox ON events; +CREATE TRIGGER events_workflow_outbox +AFTER INSERT ON events +FOR EACH ROW EXECUTE FUNCTION enqueue_workflow_event_outbox(); diff --git a/schema/schema.sql b/schema/schema.sql index f5f32cc3e3..cdc94f9d68 100644 --- a/schema/schema.sql +++ b/schema/schema.sql @@ -376,6 +376,8 @@ CREATE TABLE workflow_runs ( community_id UUID NOT NULL REFERENCES communities(id), id UUID NOT NULL DEFAULT gen_random_uuid(), workflow_id UUID NOT NULL, + definition_snapshot JSONB, + definition_hash BYTEA, status run_status NOT NULL DEFAULT 'pending', trigger_event_id BYTEA, current_step INT NOT NULL DEFAULT 0, @@ -384,6 +386,13 @@ CREATE TABLE workflow_runs ( started_at TIMESTAMPTZ, completed_at TIMESTAMPTZ, error_message TEXT, + lease_owner TEXT, + lease_token UUID, + lease_expires_at TIMESTAMPTZ, + heartbeat_at TIMESTAMPTZ, + attempt INTEGER NOT NULL DEFAULT 0, + available_at TIMESTAMPTZ NOT NULL DEFAULT NOW(), + recovery_classification TEXT, created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(), PRIMARY KEY (community_id, id), FOREIGN KEY (community_id, workflow_id) @@ -392,6 +401,57 @@ CREATE TABLE workflow_runs ( CREATE INDEX idx_workflow_runs_workflow ON workflow_runs (community_id, workflow_id); CREATE INDEX idx_workflow_runs_status ON workflow_runs (community_id, status); +CREATE INDEX idx_workflow_runs_claimable ON workflow_runs (community_id, available_at, created_at) + WHERE status IN ('pending', 'waiting_approval', 'failed', 'running'); +CREATE INDEX idx_workflow_runs_lease_expiry ON workflow_runs (lease_expires_at) + WHERE status = 'running'; +CREATE UNIQUE INDEX uq_workflow_runs_trigger_identity + ON workflow_runs (community_id, workflow_id, trigger_event_id) + WHERE trigger_event_id IS NOT NULL; + +CREATE TABLE workflow_step_attempts ( + community_id UUID NOT NULL, + run_id UUID NOT NULL, + step_index INTEGER NOT NULL, + step_id VARCHAR(64) NOT NULL, + attempt INTEGER NOT NULL, + idempotency_key TEXT NOT NULL, + status TEXT NOT NULL, + started_at TIMESTAMPTZ NOT NULL DEFAULT NOW(), + completed_at TIMESTAMPTZ, + result JSONB, + error_message TEXT, + recovery_classification TEXT, + PRIMARY KEY (community_id, run_id, step_index, attempt), + UNIQUE (community_id, idempotency_key), + FOREIGN KEY (community_id, run_id) + REFERENCES workflow_runs (community_id, id) ON DELETE CASCADE +); +CREATE INDEX idx_workflow_step_attempts_recovery + ON workflow_step_attempts (community_id, status, started_at); +CREATE TABLE workflow_event_outbox ( + community_id UUID NOT NULL, + event_id BYTEA NOT NULL, + available_at TIMESTAMPTZ NOT NULL DEFAULT NOW(), + attempts INTEGER NOT NULL DEFAULT 0, + claimed_by TEXT, + claimed_until TIMESTAMPTZ, + created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(), + PRIMARY KEY (community_id, event_id) +); +CREATE INDEX idx_workflow_event_outbox_claimable + ON workflow_event_outbox (available_at, claimed_until); +CREATE TABLE workflow_action_effects ( + community_id UUID NOT NULL, + run_id UUID NOT NULL, + idempotency_key TEXT NOT NULL, + event_id BYTEA, + status TEXT NOT NULL DEFAULT 'reserved', + created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(), + PRIMARY KEY (community_id, idempotency_key), + FOREIGN KEY (community_id, run_id) + REFERENCES workflow_runs (community_id, id) ON DELETE CASCADE +); -- ── Workflow approvals ──────────────────────────────────────────────────────── -- token-hash lookup scoped: approval token grants cannot act on another @@ -935,6 +995,28 @@ AFTER INSERT ON events DEFERRABLE INITIALLY DEFERRED FOR EACH ROW EXECUTE FUNCTION refresh_channel_ttl_after_event_insert(); +CREATE OR REPLACE FUNCTION enqueue_workflow_event_outbox() RETURNS trigger +LANGUAGE plpgsql AS $$ +BEGIN + IF NEW.channel_id IS NOT NULL AND EXISTS ( + SELECT 1 + FROM workflows w + WHERE w.community_id = NEW.community_id + AND w.channel_id = NEW.channel_id + AND w.status = 'active' + AND w.enabled + ) THEN + INSERT INTO workflow_event_outbox (community_id, event_id) + VALUES (NEW.community_id, NEW.id) + ON CONFLICT (community_id, event_id) DO NOTHING; + END IF; + RETURN NEW; +END +$$; +CREATE TRIGGER events_workflow_outbox +AFTER INSERT ON events +FOR EACH ROW EXECUTE FUNCTION enqueue_workflow_event_outbox(); + -- Replica-fence floor guard (keep in sync with migrations/0021). A deferred -- constraint trigger re-checks, inside COMMIT processing, that channel-bearing -- event rows are no older than `buzz.created_at_floor` seconds before commit