diff --git a/crates/buzz-db/src/lib.rs b/crates/buzz-db/src/lib.rs index 9b26876747..9579e5f04f 100644 --- a/crates/buzz-db/src/lib.rs +++ b/crates/buzz-db/src/lib.rs @@ -3585,6 +3585,35 @@ impl Db { .await } + /// Insert or update a workflow inside an existing transaction. + /// + /// Used by relay event ingest so the signed kind-30620 source event and + /// the workflow registry projection share one commit boundary. + #[allow(clippy::too_many_arguments)] + pub async fn upsert_workflow_tx( + &self, + tx: &mut sqlx::Transaction<'_, sqlx::Postgres>, + community_id: CommunityId, + id: Uuid, + channel_id: Option, + owner_pubkey: &[u8], + name: &str, + definition_json: &str, + definition_hash: &[u8], + ) -> Result<()> { + workflow::upsert_workflow_tx( + tx, + community_id, + id, + channel_id, + owner_pubkey, + name, + definition_json, + definition_hash, + ) + .await + } + /// Fetch a single workflow by ID, scoped to its community. pub async fn get_workflow( &self, diff --git a/crates/buzz-db/src/workflow.rs b/crates/buzz-db/src/workflow.rs index 7a2396c1fd..512012a49d 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::{Executor, PgPool, Postgres, Row, Transaction}; use uuid::Uuid; use buzz_core::CommunityId; @@ -320,6 +320,62 @@ pub async fn upsert_workflow( definition_json: &str, definition_hash: &[u8], ) -> Result<()> { + upsert_workflow_with_executor( + pool, + community_id, + id, + channel_id, + owner_pubkey, + name, + definition_json, + definition_hash, + ) + .await +} + +/// Insert or update a workflow inside an existing transaction. +/// +/// The relay uses this variant to commit the signed kind-30620 event and its +/// executable registry projection together. A rollback must leave neither +/// representation behind. +#[allow(clippy::too_many_arguments)] +pub async fn upsert_workflow_tx( + tx: &mut Transaction<'_, Postgres>, + community_id: CommunityId, + id: Uuid, + channel_id: Option, + owner_pubkey: &[u8], + name: &str, + definition_json: &str, + definition_hash: &[u8], +) -> Result<()> { + upsert_workflow_with_executor( + &mut **tx, + community_id, + id, + channel_id, + owner_pubkey, + name, + definition_json, + definition_hash, + ) + .await +} + +#[allow(clippy::too_many_arguments)] +async fn upsert_workflow_with_executor<'e, E>( + executor: E, + community_id: CommunityId, + id: Uuid, + channel_id: Option, + owner_pubkey: &[u8], + name: &str, + definition_json: &str, + definition_hash: &[u8], +) -> Result<()> +where + E: Executor<'e, Database = Postgres>, +{ let row = sqlx::query( r#" INSERT INTO workflows @@ -342,7 +398,7 @@ pub async fn upsert_workflow( .bind(channel_id) .bind(definition_json) .bind(definition_hash) - .fetch_optional(pool) + .fetch_optional(executor) .await?; if row.is_none() { @@ -1770,6 +1826,153 @@ mod tests { id } + /// A failed relay ingest must not leave an executable registry row without + /// its signed kind-30620 source event. The relay now writes both through one + /// transaction; this pins the registry half of that rollback boundary. + #[tokio::test] + #[ignore = "requires Postgres"] + async fn transactional_upsert_rolls_back_registry_projection() { + let pool = setup_pool().await; + let community = make_community(&pool).await; + let owner = vec![0xb3; 32]; + ensure_user(&pool, community, &owner) + .await + .expect("ensure owner"); + let channel_id = make_channel(&pool, community, &owner).await; + let workflow_id = Uuid::new_v4(); + + let mut tx = pool.begin().await.expect("begin transaction"); + upsert_workflow_tx( + &mut tx, + community, + workflow_id, + Some(channel_id), + &owner, + "atomic projection", + r#"{"name":"atomic projection","trigger":{"on":"schedule"},"steps":[]}"#, + &[0x44; 32], + ) + .await + .expect("transactional workflow upsert"); + + let inside_tx: i64 = sqlx::query_scalar( + "SELECT COUNT(*) FROM workflows WHERE community_id = $1 AND id = $2", + ) + .bind(community.as_uuid()) + .bind(workflow_id) + .fetch_one(&mut *tx) + .await + .expect("read uncommitted workflow in its transaction"); + assert_eq!(inside_tx, 1); + + tx.rollback().await.expect("rollback transaction"); + + let after_rollback: i64 = sqlx::query_scalar( + "SELECT COUNT(*) FROM workflows WHERE community_id = $1 AND id = $2", + ) + .bind(community.as_uuid()) + .bind(workflow_id) + .fetch_one(&pool) + .await + .expect("read workflow after rollback"); + assert_eq!( + after_rollback, 0, + "rollback must not leave a registry-only workflow projection" + ); + } + + #[tokio::test] + #[ignore = "requires Postgres"] + async fn transactional_upsert_is_idempotent_and_rejects_owner_or_channel_changes() { + let pool = setup_pool().await; + let community = make_community(&pool).await; + let owner = vec![0xb4; 32]; + let other_owner = vec![0xb5; 32]; + ensure_user(&pool, community, &owner) + .await + .expect("ensure owner"); + ensure_user(&pool, community, &other_owner) + .await + .expect("ensure other owner"); + let channel = make_channel(&pool, community, &owner).await; + let other_channel = make_channel(&pool, community, &owner).await; + let workflow_id = Uuid::new_v4(); + let initial = r#"{"name":"initial","trigger":{"on":"schedule"},"steps":[]}"#; + let updated = r#"{"name":"updated","trigger":{"on":"schedule"},"steps":[]}"#; + + let mut create_tx = pool.begin().await.expect("begin create transaction"); + upsert_workflow_tx( + &mut create_tx, + community, + workflow_id, + Some(channel), + &owner, + "initial", + initial, + &[0x45; 32], + ) + .await + .expect("initial transactional upsert"); + create_tx.commit().await.expect("commit initial workflow"); + + let mut retry_tx = pool.begin().await.expect("begin retry transaction"); + upsert_workflow_tx( + &mut retry_tx, + community, + workflow_id, + Some(channel), + &owner, + "updated", + updated, + &[0x46; 32], + ) + .await + .expect("same owner/channel retry"); + retry_tx.commit().await.expect("commit retry"); + + let stored: (i64, String) = sqlx::query_as( + "SELECT COUNT(*) OVER(), name FROM workflows WHERE community_id = $1 AND id = $2", + ) + .bind(community.as_uuid()) + .bind(workflow_id) + .fetch_one(&pool) + .await + .expect("read idempotently updated workflow"); + assert_eq!(stored, (1, "updated".to_string())); + + let mut owner_tx = pool.begin().await.expect("begin wrong-owner transaction"); + let owner_error = upsert_workflow_tx( + &mut owner_tx, + community, + workflow_id, + Some(channel), + &other_owner, + "stolen", + updated, + &[0x47; 32], + ) + .await + .expect_err("different owner must be rejected"); + assert!(matches!(owner_error, DbError::AccessDenied(_))); + owner_tx.rollback().await.expect("rollback wrong owner"); + + let mut channel_tx = pool.begin().await.expect("begin wrong-channel transaction"); + let channel_error = upsert_workflow_tx( + &mut channel_tx, + community, + workflow_id, + Some(other_channel), + &owner, + "moved", + updated, + &[0x48; 32], + ) + .await + .expect_err("different channel must be rejected"); + assert!(matches!(channel_error, DbError::AccessDenied(_))); + channel_tx.rollback().await.expect("rollback wrong channel"); + } + /// Insert a workflow whose tenant is `community`'s channel. Returns the /// workflow id and the owning community for callers that want to assert /// the resolved tenant. diff --git a/crates/buzz-relay/src/handlers/command_executor.rs b/crates/buzz-relay/src/handlers/command_executor.rs index 2d82736807..405f1fd9f2 100644 --- a/crates/buzz-relay/src/handlers/command_executor.rs +++ b/crates/buzz-relay/src/handlers/command_executor.rs @@ -91,12 +91,14 @@ enum PersistResult { /// If the event is a duplicate (ON CONFLICT DO NOTHING), the transaction is /// rolled back and `PersistResult::Duplicate` is returned — no mutations needed. /// -/// NOTE: Domain mutations (open_dm, upsert_workflow, etc.) execute on the +/// NOTE: Most domain mutations (open_dm, hide_dm, approvals) execute on the /// connection pool, NOT inside this transaction. The pattern is idempotent but /// not strictly atomic: if a mutation succeeds but commit fails, the mutation /// persists without the event record. On retry, the event INSERT succeeds /// (no conflict), and the mutation re-executes — which is safe for idempotent -/// operations (open_dm, hide_dm, update_approval, upsert_workflow). +/// operations. Workflow definitions are the exception: their registry upsert +/// uses this same transaction so the signed kind-30620 event and executable +/// projection can never diverge. async fn persist_command_event( state: &Arc, tenant: &TenantContext, @@ -749,8 +751,9 @@ async fn handle_workflow_def( .map_err(|e| IngestError::Internal(format!("error: json serialize: {e}")))?; let hash = compute_definition_hash(&definition_json_final); - // Persist the command event — returns open transaction - let tx = match persist_command_event(state, tenant, event, None).await? { + // Persist the command event — returns open transaction. The workflow + // registry projection is written through this same transaction below. + let mut tx = match persist_command_event(state, tenant, event, None).await? { PersistResult::Duplicate => { return Ok(IngestResult { event_id: event.id.to_hex(), @@ -781,7 +784,8 @@ async fn handle_workflow_def( state .db - .upsert_workflow( + .upsert_workflow_tx( + &mut tx, community_id, workflow_id, Some(channel_id), @@ -1368,3 +1372,156 @@ async fn resume_workflow_after_approval( .finalize_run(community_id, run_id, result, existing_trace) .await; } + +#[cfg(test)] +mod integration_tests { + //! Postgres-gated regression for the workflow definition atomicity boundary. + //! + //! Run with: + //! `cargo test -p buzz-relay --lib workflow_definition_event_and_registry_roll_back_together -- --ignored --exact` + + use super::*; + use buzz_core::channel::{ChannelType, ChannelVisibility}; + use buzz_db::CreateCommunityWithOwnerResult; + use nostr::Keys; + use sqlx::PgPool; + + async fn test_state() -> (Arc, PgPool) { + let mut config = crate::config::Config::from_env().expect("default config loads"); + config.require_relay_membership = false; + config.redis_url = "redis://127.0.0.1:1".to_string(); + let pool = PgPool::connect_lazy(&config.database_url).expect("lazy pg pool"); + let db = buzz_db::Db::from_pool(pool.clone()); + let redis_pool = deadpool_redis::Config::from_url(&config.redis_url) + .create_pool(Some(deadpool_redis::Runtime::Tokio1)) + .expect("redis pool"); + let pubsub = Arc::new( + buzz_pubsub::PubSubManager::new(&config.redis_url, redis_pool.clone()) + .await + .expect("pubsub manager"), + ); + let audit = buzz_audit::AuditService::new(pool.clone()); + let auth = buzz_auth::AuthService::new(config.auth.clone()); + let search = buzz_search::SearchService::new(pool.clone()); + let workflow_engine = Arc::new(buzz_workflow::WorkflowEngine::new( + db.clone(), + buzz_workflow::WorkflowConfig::default(), + )); + let media_storage = buzz_media::MediaStorage::new(&config.media).expect("media storage"); + let (state, _audit_shutdown) = AppState::new( + config, + db, + redis_pool, + audit, + pubsub, + auth, + search, + workflow_engine, + Keys::generate(), + media_storage, + ); + (Arc::new(state), pool) + } + + #[tokio::test] + #[ignore = "requires Postgres"] + async fn workflow_definition_event_and_registry_roll_back_together() { + let (state, pool) = test_state().await; + let owner = Keys::generate(); + let owner_hex = owner.public_key().to_hex(); + let host = format!("wf-atomic-{}.example", Uuid::new_v4().simple()); + let community = match state + .db + .create_community_with_owner(&host, &owner_hex) + .await + .expect("create community") + { + CreateCommunityWithOwnerResult::Created(record) => record.id, + other => panic!("expected fresh community, got {other:?}"), + }; + let channel = state + .db + .create_channel( + community, + "workflow-atomicity", + ChannelType::Stream, + ChannelVisibility::Open, + None, + &owner.public_key().to_bytes(), + None, + ) + .await + .expect("create channel"); + let tenant = TenantContext::resolved(community, host); + let workflow_id = Uuid::new_v4(); + let yaml = "name: atomic projection\ntrigger:\n on: schedule\nsteps: []\n"; + let event = buzz_sdk::builders::build_workflow_def(channel.id, workflow_id, yaml) + .expect("build workflow definition") + .sign_with_keys(&owner) + .expect("sign workflow definition"); + + let mut tx = match persist_command_event(&state, &tenant, &event, None) + .await + .expect("persist signed event") + { + PersistResult::Inserted(tx) => tx, + PersistResult::Duplicate => panic!("fresh event must be inserted"), + }; + let definition = r#"{"name":"atomic projection","trigger":{"on":"schedule"},"steps":[]}"#; + let hash = compute_definition_hash(definition); + state + .db + .upsert_workflow_tx( + &mut tx, + community, + workflow_id, + Some(channel.id), + &owner.public_key().to_bytes(), + "atomic projection", + definition, + &hash, + ) + .await + .expect("upsert registry projection"); + + let event_inside: i64 = + sqlx::query_scalar("SELECT COUNT(*) FROM events WHERE community_id = $1 AND id = $2") + .bind(community.as_uuid()) + .bind(event.id.as_bytes().as_slice()) + .fetch_one(tx.as_mut()) + .await + .expect("read uncommitted event"); + let workflow_inside: i64 = sqlx::query_scalar( + "SELECT COUNT(*) FROM workflows WHERE community_id = $1 AND id = $2", + ) + .bind(community.as_uuid()) + .bind(workflow_id) + .fetch_one(tx.as_mut()) + .await + .expect("read uncommitted workflow"); + assert_eq!((event_inside, workflow_inside), (1, 1)); + + tx.rollback().await.expect("force ingest rollback"); + + let event_after: i64 = + sqlx::query_scalar("SELECT COUNT(*) FROM events WHERE community_id = $1 AND id = $2") + .bind(community.as_uuid()) + .bind(event.id.as_bytes().as_slice()) + .fetch_one(&pool) + .await + .expect("read event after rollback"); + let workflow_after: i64 = sqlx::query_scalar( + "SELECT COUNT(*) FROM workflows WHERE community_id = $1 AND id = $2", + ) + .bind(community.as_uuid()) + .bind(workflow_id) + .fetch_one(&pool) + .await + .expect("read workflow after rollback"); + assert_eq!( + (event_after, workflow_after), + (0, 0), + "rollback must remove both the signed event and registry projection" + ); + } +} diff --git a/crates/buzz-workflow/src/lib.rs b/crates/buzz-workflow/src/lib.rs index 7aaa3d1702..e1a6071251 100644 --- a/crates/buzz-workflow/src/lib.rs +++ b/crates/buzz-workflow/src/lib.rs @@ -53,6 +53,13 @@ use dashmap::DashMap; use tokio::sync::Semaphore; use uuid::Uuid; +/// The scheduler sleeps for 60 seconds between scans, but each scan also does +/// database and workflow work. A lookback equal to the sleep duration leaves a +/// gap whenever that work pushes the next scan even slightly past 60 seconds. +/// Keep one extra tick of overlap; durable scheduled-fire claims deduplicate +/// the overlap across ticks and pods. +const CRON_LOOKBACK_SECS: i64 = 120; + /// Runtime configuration for the workflow engine. #[derive(Clone, Debug)] pub struct WorkflowConfig { @@ -536,7 +543,7 @@ impl WorkflowEngine { schema::TriggerDef::Schedule { cron: Some(expr), interval: None, - } => match cron_fire_instant(expr, now, 60, workflow.id) { + } => match cron_fire_instant(expr, now, CRON_LOOKBACK_SECS, workflow.id) { Some(instant) => (instant, "cron"), None => continue, }, @@ -746,10 +753,11 @@ impl WorkflowEngine { /// Find the cron schedule instant that fired within the `window_secs`-wide /// window ending at `now`, if any. /// -/// Uses window-based matching: finds the next scheduled time after -/// `(now - window_secs)` and returns it when it falls at or before `now`. -/// This tolerates tick drift gracefully — a 61s tick won't miss a -/// minute-granularity cron expression. The returned instant is the cron's own +/// Uses window-based matching: finds scheduled times after +/// `(now - window_secs)` and returns the latest one at or before `now`. +/// Returning the latest due instant matters when an overlapping lookback spans +/// multiple minute-granularity fires: each tick should claim the freshest one, +/// not remain one fire behind. The returned instant is the cron's own /// scheduled time (not `now`), so every pod evaluating the same expression in /// the same window computes the *same* value — making it a safe, deterministic /// claim anchor for cross-pod at-most-once firing. @@ -766,7 +774,10 @@ fn cron_fire_instant( match normalized.parse::() { Ok(sched) => { let window_start = now - chrono::Duration::seconds(window_secs); - sched.after(&window_start).next().filter(|t| *t <= now) + sched + .after(&window_start) + .take_while(|scheduled| *scheduled <= now) + .last() } Err(e) => { tracing::warn!( @@ -1136,6 +1147,44 @@ mod tests { ); } + #[test] + fn cron_fire_instant_survives_tick_work_overrun() { + // The loop sleeps 60s *after* doing work, so a scan can arrive just + // beyond a one-minute boundary. The production lookback overlaps one + // additional tick and must still recover the scheduled fire. + let now = chrono::DateTime::parse_from_rfc3339("2026-06-15T09:01:01Z") + .unwrap() + .with_timezone(&Utc); + let scheduled = chrono::DateTime::parse_from_rfc3339("2026-06-15T09:00:00Z") + .unwrap() + .with_timezone(&Utc); + let wf_id = Uuid::new_v4(); + assert_eq!( + cron_fire_instant("0 9 * * *", now, CRON_LOOKBACK_SECS, wf_id), + Some(scheduled), + "scheduler work must not open a gap immediately after the 60s sleep" + ); + } + + #[test] + fn cron_fire_instant_returns_latest_due_fire_in_overlapping_window() { + // A two-minute lookback spans two fires for an every-minute workflow. + // Choose the current minute; choosing the oldest would keep each tick + // permanently one fire behind. + let now = chrono::DateTime::parse_from_rfc3339("2026-06-15T12:02:30Z") + .unwrap() + .with_timezone(&Utc); + let latest = chrono::DateTime::parse_from_rfc3339("2026-06-15T12:02:00Z") + .unwrap() + .with_timezone(&Utc); + let wf_id = Uuid::new_v4(); + assert_eq!( + cron_fire_instant("* * * * *", now, CRON_LOOKBACK_SECS, wf_id), + Some(latest), + "overlap must claim the latest due schedule instant" + ); + } + #[test] fn cron_fire_instant_returns_none_just_outside_window() { // Fixed time: 09:01:01 UTC. Cron "0 9 * * *" fires at 09:00:00.