From 72f92568e811745d85700e39f46cd68a5ca7bfac Mon Sep 17 00:00:00 2001 From: Logan Johnson Date: Thu, 27 Aug 2026 17:07:36 -0400 Subject: [PATCH] Persist exact workflow revisions additively Signed-off-by: Logan Johnson Co-authored-by: Larry <8cf5a83f590ec0955b11647d1c88f796a98e088c30a492c58e0e46c3026ae7a4@buzz.block.builderlab.xyz> Signed-off-by: Larry <8cf5a83f590ec0955b11647d1c88f796a98e088c30a492c58e0e46c3026ae7a4@buzz.block.builderlab.xyz> Signed-off-by: Logan Johnson --- crates/buzz-db/src/runtime/migration.rs | 18 +++ crates/buzz-db/src/store/workflow.rs | 145 +++++++++++++++--- crates/buzz-relay/src/api/bridge.rs | 8 +- .../src/handlers/command_executor.rs | 30 ++-- crates/buzz-workflow/src/lib.rs | 2 + migrations/0040_workflow_revision_binding.sql | 8 + schema/schema.sql | 6 + 7 files changed, 183 insertions(+), 34 deletions(-) create mode 100644 migrations/0040_workflow_revision_binding.sql diff --git a/crates/buzz-db/src/runtime/migration.rs b/crates/buzz-db/src/runtime/migration.rs index 5901512504..baad0b3e30 100644 --- a/crates/buzz-db/src/runtime/migration.rs +++ b/crates/buzz-db/src/runtime/migration.rs @@ -1165,6 +1165,24 @@ mod postgres_tests { assert!(heartbeat_vacuum.contains("vacuum_truncate = false")); assert!(desired_schema.contains("vacuum_truncate = false")); + // Workflow revision capture is additive: nullable 32-byte event IDs on + // both the materialized definition and run, with identical fresh-schema + // constraints and no backfill hidden in startup migration state. + assert_eq!(migrations[39].version, 40); + let workflow_revision_binding = migrations[39].sql.as_str(); + assert!(workflow_revision_binding.contains("ALTER TABLE workflows")); + assert!(workflow_revision_binding.contains("ALTER TABLE workflow_runs")); + assert!(workflow_revision_binding.contains("octet_length(definition_event_id) = 32")); + assert!(!workflow_revision_binding + .to_ascii_lowercase() + .contains("update ")); + assert_eq!( + desired_schema + .matches("octet_length(definition_event_id) = 32") + .count(), + 2 + ); + // pgschema intentionally reconciles DDL, not seed DML or table storage // parameters. Its post-apply reconciliation must restore and verify // both parts of the live heartbeat contract for fresh bootstraps. diff --git a/crates/buzz-db/src/store/workflow.rs b/crates/buzz-db/src/store/workflow.rs index 3ceed9ea32..427e5b211b 100644 --- a/crates/buzz-db/src/store/workflow.rs +++ b/crates/buzz-db/src/store/workflow.rs @@ -179,6 +179,9 @@ pub struct WorkflowRecord { pub definition: serde_json::Value, /// SHA-256 hash of the canonical definition JSON. pub definition_hash: Vec, + /// Exact owner-signed kind:30620 event that materialized this revision. + /// NULL is retained for workflows created before revision capture. + pub definition_event_id: Option>, /// Current lifecycle status of the workflow definition. pub status: WorkflowStatus, /// Whether the workflow will fire on matching events. @@ -203,6 +206,9 @@ pub struct WorkflowRunRecord { pub community_id: CommunityId, /// The workflow definition that was executed. pub workflow_id: Uuid, + /// Exact owner-signed kind:30620 revision selected when this run was created. + /// NULL is retained when the workflow has no captured revision yet. + pub definition_event_id: Option>, /// Current execution status of this run. pub status: RunStatus, /// Raw event ID bytes that triggered this run, if any. @@ -316,7 +322,7 @@ pub async fn create_workflow( /// cross-channel overwrite primitive while still making retries idempotent. #[allow(clippy::too_many_arguments)] pub async fn upsert_workflow( - pool: &PgPool, + tx: &mut sqlx::Transaction<'_, sqlx::Postgres>, community_id: CommunityId, id: Uuid, channel_id: Option, @@ -324,16 +330,18 @@ pub async fn upsert_workflow( name: &str, definition_json: &str, definition_hash: &[u8], + definition_event_id: &[u8], ) -> Result<()> { let row = sqlx::query( r#" INSERT INTO workflows - (community_id, id, name, owner_pubkey, channel_id, definition, definition_hash, status, enabled) - VALUES ($1, $2, $3, $4, $5, $6::jsonb, $7, 'active', TRUE) + (community_id, id, name, owner_pubkey, channel_id, definition, definition_hash, definition_event_id, status, enabled) + VALUES ($1, $2, $3, $4, $5, $6::jsonb, $7, $8, 'active', TRUE) ON CONFLICT (community_id, id) DO UPDATE SET name = EXCLUDED.name, definition = EXCLUDED.definition, definition_hash = EXCLUDED.definition_hash, + definition_event_id = EXCLUDED.definition_event_id, updated_at = NOW() WHERE workflows.owner_pubkey = EXCLUDED.owner_pubkey AND workflows.channel_id IS NOT DISTINCT FROM EXCLUDED.channel_id @@ -347,7 +355,8 @@ pub async fn upsert_workflow( .bind(channel_id) .bind(definition_json) .bind(definition_hash) - .fetch_optional(pool) + .bind(definition_event_id) + .fetch_optional(&mut **tx) .await?; if row.is_none() { @@ -372,7 +381,7 @@ pub async fn get_workflow( ) -> Result { let row = sqlx::query( r#" - SELECT id, community_id, name, owner_pubkey, channel_id, definition, definition_hash, + SELECT id, community_id, name, owner_pubkey, channel_id, definition, definition_hash, definition_event_id, status::text AS status, enabled, created_at, updated_at FROM workflows WHERE community_id = $1 AND id = $2 @@ -403,7 +412,7 @@ pub async fn list_channel_workflows( let rows = sqlx::query( r#" - SELECT id, community_id, name, owner_pubkey, channel_id, definition, definition_hash, + SELECT id, community_id, name, owner_pubkey, channel_id, definition, definition_hash, definition_event_id, status::text AS status, enabled, created_at, updated_at FROM workflows WHERE community_id = $1 AND channel_id = $2 @@ -434,7 +443,7 @@ pub async fn list_enabled_channel_workflows( ) -> Result> { let rows = sqlx::query( r#" - SELECT id, community_id, name, owner_pubkey, channel_id, definition, definition_hash, + SELECT id, community_id, name, owner_pubkey, channel_id, definition, definition_hash, definition_event_id, status::text AS status, enabled, created_at, updated_at FROM workflows WHERE community_id = $1 @@ -462,7 +471,7 @@ pub async fn list_enabled_channel_workflows( pub async fn list_all_enabled_workflows(pool: &PgPool) -> Result> { let rows = sqlx::query( r#" - SELECT w.id, w.community_id, w.name, w.owner_pubkey, w.channel_id, w.definition, w.definition_hash, + SELECT w.id, w.community_id, w.name, w.owner_pubkey, w.channel_id, w.definition, w.definition_hash, w.definition_event_id, w.status::text AS status, w.enabled, w.created_at, w.updated_at FROM workflows w JOIN communities c ON c.id = w.community_id @@ -804,6 +813,7 @@ pub async fn create_workflow_run( pool: &PgPool, community_id: CommunityId, workflow_id: Uuid, + definition_event_id: Option<&[u8]>, trigger_event_id: Option<&[u8]>, trigger_context: Option<&serde_json::Value>, ) -> Result { @@ -812,13 +822,14 @@ pub async fn create_workflow_run( 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_event_id, status, trigger_event_id, current_step, execution_trace, trigger_context) + VALUES ($1, $2, $3, $4, 'pending', $5, 0, '[]', $6) "#, ) .bind(community_id.as_uuid()) .bind(id) .bind(workflow_id) + .bind(definition_event_id) .bind(trigger_event_id) .bind(trigger_context) .execute(pool) @@ -835,7 +846,7 @@ pub async fn get_workflow_run( ) -> Result { let row = sqlx::query( r#" - SELECT community_id, id, workflow_id, status::text AS status, trigger_event_id, current_step, + SELECT community_id, id, workflow_id, definition_event_id, status::text AS status, trigger_event_id, current_step, execution_trace, trigger_context, started_at, completed_at, error_message, error_code, created_at FROM workflow_runs WHERE community_id = $1 AND id = $2 @@ -867,7 +878,7 @@ pub async fn list_workflow_runs_page( let limit = limit.clamp(1, LIST_MAX_LIMIT); let rows = sqlx::query( r#" - SELECT community_id, id, workflow_id, status::text AS status, trigger_event_id, current_step, + SELECT community_id, id, workflow_id, definition_event_id, status::text AS status, trigger_event_id, current_step, execution_trace, trigger_context, started_at, completed_at, error_message, error_code, created_at FROM workflow_runs WHERE community_id = $1 AND workflow_id = $2 @@ -1185,6 +1196,7 @@ fn row_to_workflow_record(row: sqlx::postgres::PgRow) -> Result channel_id, definition: row.try_get("definition")?, definition_hash: row.try_get("definition_hash")?, + definition_event_id: row.try_get("definition_event_id")?, status, enabled, created_at: row.try_get("created_at")?, @@ -1204,6 +1216,7 @@ fn row_to_run_record(row: sqlx::postgres::PgRow) -> Result { id, community_id: CommunityId::from_uuid(community_id), workflow_id, + definition_event_id: row.try_get("definition_event_id")?, status, trigger_event_id: row.try_get("trigger_event_id")?, current_step: row.try_get("current_step")?, @@ -1249,7 +1262,7 @@ pub async fn find_by_owner_and_name( ) -> Result> { let row = sqlx::query( r#" - SELECT id, community_id, name, owner_pubkey, channel_id, definition, definition_hash, + SELECT id, community_id, name, owner_pubkey, channel_id, definition, definition_hash, definition_event_id, status::text AS status, enabled, created_at, updated_at FROM workflows WHERE community_id = $1 AND owner_pubkey = $2 AND name = $3 @@ -1277,6 +1290,7 @@ impl Db { &self, community_id: CommunityId, workflow_id: Uuid, + definition_event_id: Option<&[u8]>, trigger_event_id: Option<&[u8]>, trigger_context: Option<&serde_json::Value>, ) -> Result { @@ -1284,6 +1298,7 @@ impl Db { &self.pool, community_id, workflow_id, + definition_event_id, trigger_event_id, trigger_context, ) @@ -1460,6 +1475,7 @@ impl Db { name, definition_json, definition_hash, + definition_event_id, ) .await } @@ -1469,6 +1485,7 @@ impl Db { #[datastore_span(name = "upsert_workflow", system = "postgresql")] pub async fn upsert_workflow( &self, + tx: &mut sqlx::Transaction<'_, sqlx::Postgres>, community_id: CommunityId, id: Uuid, channel_id: Option, @@ -1476,9 +1493,10 @@ impl Db { name: &str, definition_json: &str, definition_hash: &[u8], + definition_event_id: &[u8], ) -> Result<()> { crate::workflow::upsert_workflow( - &self.pool, + tx, community_id, id, channel_id, @@ -1486,6 +1504,7 @@ impl Db { name, definition_json, definition_hash, + definition_event_id, ) .await } @@ -1608,6 +1627,7 @@ impl Db { name, definition_json, definition_hash, + definition_event_id, ) .await } @@ -1798,6 +1818,7 @@ mod postgres_tests { owner_pubkey: vec![0xab; 32], channel_id: Some(channel_id), definition: def.clone(), + definition_event_id: None, definition_hash: vec![0x01, 0x02, 0x03, 0x04], status: WorkflowStatus::Active, enabled: true, @@ -1828,6 +1849,7 @@ mod postgres_tests { owner_pubkey: vec![0x00; 32], channel_id: None, definition: serde_json::json!({}), + definition_event_id: None, definition_hash: vec![], status: WorkflowStatus::Active, enabled: true, @@ -1850,6 +1872,7 @@ mod postgres_tests { owner_pubkey: vec![0x01; 32], channel_id: None, definition: serde_json::json!({}), + definition_event_id: None, definition_hash: vec![0xAA], status: WorkflowStatus::Active, enabled: true, @@ -1879,6 +1902,7 @@ mod postgres_tests { owner_pubkey: vec![], channel_id: None, definition: serde_json::json!({}), + definition_event_id: None, definition_hash: vec![], status: status.clone(), enabled: true, @@ -1899,6 +1923,7 @@ mod postgres_tests { owner_pubkey: vec![], channel_id: None, definition: serde_json::json!({}), + definition_event_id: None, definition_hash: vec![], status: WorkflowStatus::Active, enabled: false, @@ -1922,6 +1947,7 @@ mod postgres_tests { id, community_id: CommunityId::from_uuid(Uuid::new_v4()), workflow_id, + definition_event_id: None, status: RunStatus::Running, trigger_event_id: Some(trigger_event_id.clone()), current_step: 2, @@ -1953,6 +1979,7 @@ mod postgres_tests { id: Uuid::new_v4(), community_id: CommunityId::from_uuid(Uuid::new_v4()), workflow_id: Uuid::new_v4(), + definition_event_id: None, status: RunStatus::Pending, trigger_event_id: None, current_step: 0, @@ -1977,6 +2004,7 @@ mod postgres_tests { id: Uuid::new_v4(), community_id: CommunityId::from_uuid(Uuid::new_v4()), workflow_id: Uuid::new_v4(), + definition_event_id: None, status: RunStatus::Failed, trigger_event_id: None, current_step: 1, @@ -2009,6 +2037,7 @@ mod postgres_tests { id: Uuid::new_v4(), community_id: CommunityId::from_uuid(Uuid::new_v4()), workflow_id: Uuid::new_v4(), + definition_event_id: None, status: RunStatus::Completed, trigger_event_id: None, current_step: 2, @@ -2032,6 +2061,7 @@ mod postgres_tests { id: Uuid::new_v4(), community_id: CommunityId::from_uuid(Uuid::new_v4()), workflow_id: Uuid::new_v4(), + definition_event_id: None, status: RunStatus::Pending, trigger_event_id: None, current_step: 0, @@ -2258,6 +2288,85 @@ mod postgres_tests { (workflow_id, community) } + #[tokio::test] + #[ignore = "requires Postgres"] + async fn revision_capture_is_additive_and_transactional() { + let pool = setup_pool().await; + let community = make_community(&pool).await; + let (workflow_id, _) = make_workflow_in(&pool, community).await; + let legacy = get_workflow(&pool, community, workflow_id) + .await + .expect("read legacy workflow"); + assert!(legacy.definition_event_id.is_none()); + + let revision = [0x42; 32]; + let legacy_run = create_workflow_run(&pool, community, workflow_id, None, None, None) + .await + .expect("create compatible legacy run"); + let bound_run = + create_workflow_run(&pool, community, workflow_id, Some(&revision), None, None) + .await + .expect("create revision-bound run"); + assert!(get_workflow_run(&pool, community, legacy_run) + .await + .expect("read legacy run") + .definition_event_id + .is_none()); + assert_eq!( + get_workflow_run(&pool, community, bound_run) + .await + .expect("read bound run") + .definition_event_id + .as_deref(), + Some(revision.as_slice()) + ); + + let mut tx = pool.begin().await.expect("begin revision update"); + upsert_workflow( + &mut tx, + community, + workflow_id, + legacy.channel_id, + &legacy.owner_pubkey, + &legacy.name, + &legacy.definition.to_string(), + &legacy.definition_hash, + &revision, + ) + .await + .expect("capture revision"); + tx.rollback().await.expect("roll back revision update"); + assert!(get_workflow(&pool, community, workflow_id) + .await + .expect("read rolled-back workflow") + .definition_event_id + .is_none()); + + let mut tx = pool.begin().await.expect("begin committed update"); + upsert_workflow( + &mut tx, + community, + workflow_id, + legacy.channel_id, + &legacy.owner_pubkey, + &legacy.name, + &legacy.definition.to_string(), + &legacy.definition_hash, + &revision, + ) + .await + .expect("capture committed revision"); + tx.commit().await.expect("commit revision update"); + assert_eq!( + get_workflow(&pool, community, workflow_id) + .await + .expect("read revision-bound workflow") + .definition_event_id + .as_deref(), + Some(revision.as_slice()) + ); + } + /// Confinement: a duplicate workflow UUID existing in both community A and /// community B must claim independently. Claiming `(A, id, t)` must NOT /// consume `(B, id, t)` — B's identical instant stays claimable, and the @@ -2380,7 +2489,7 @@ mod postgres_tests { .expect("claim wins"); // Create the run the won claim is responsible for, then attach it. - let run_id = create_workflow_run(&pool, community, workflow_id, None, None) + let run_id = create_workflow_run(&pool, community, workflow_id, None, None, None) .await .expect("create run ok"); @@ -2409,7 +2518,7 @@ mod postgres_tests { // A second attach is a no-op: the `workflow_run_id IS NULL` guard means // an already-linked claim is never re-pointed to a different run. - let other_run = create_workflow_run(&pool, community, workflow_id, None, None) + let other_run = create_workflow_run(&pool, community, workflow_id, None, None, None) .await .expect("create second run ok"); let reattached = @@ -2690,10 +2799,10 @@ mod postgres_tests { insert_workflow_with_ids(&pool, community_a, workflow_id, channel_id, "wf-A").await; insert_workflow_with_ids(&pool, community_b, workflow_id, Uuid::new_v4(), "wf-B").await; - let run_a = create_workflow_run(&pool, community_a, workflow_id, None, None) + let run_a = create_workflow_run(&pool, community_a, workflow_id, None, None, None) .await .expect("run A"); - let run_b = create_workflow_run(&pool, community_b, workflow_id, None, None) + let run_b = create_workflow_run(&pool, community_b, workflow_id, None, None, None) .await .expect("run B"); diff --git a/crates/buzz-relay/src/api/bridge.rs b/crates/buzz-relay/src/api/bridge.rs index 37c549610d..09174cfd17 100644 --- a/crates/buzz-relay/src/api/bridge.rs +++ b/crates/buzz-relay/src/api/bridge.rs @@ -2180,7 +2180,13 @@ pub async fn workflow_webhook( let run_id = state .db - .create_workflow_run(community_id, id, None, trigger_ctx_json.as_ref()) + .create_workflow_run( + community_id, + id, + workflow.definition_event_id.as_deref(), + None, + trigger_ctx_json.as_ref(), + ) .await .map_err(|e| super::internal_error(&format!("db error: {e}")))?; diff --git a/crates/buzz-relay/src/handlers/command_executor.rs b/crates/buzz-relay/src/handlers/command_executor.rs index 074f6b391d..48675e9019 100644 --- a/crates/buzz-relay/src/handlers/command_executor.rs +++ b/crates/buzz-relay/src/handlers/command_executor.rs @@ -92,12 +92,11 @@ 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 -/// 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). +/// Most domain mutations still execute on the connection pool rather than this +/// transaction, so their handlers rely on idempotency if event commit fails. +/// Workflow definition ingest is the exception: it materializes the workflow +/// revision on this same transaction so the signed event and revision pointer +/// commit or roll back together. #[datastore_span(name = "persist_command_event", system = "postgresql")] async fn persist_command_event( db: &buzz_db::Db, @@ -737,8 +736,8 @@ async fn handle_workflow_def( .map_err(|e| IngestError::Internal(format!("error: json serialize: {e}")))?; let hash = compute_definition_hash(&definition_json_final); - // Persist the command event — returns open transaction - let tx = match persist_command_event(&state.db, tenant, event, None).await? { + // Persist the signed definition and materialized revision atomically. + let mut tx = match persist_command_event(&state.db, tenant, event, Some(channel_id)).await? { PersistResult::Duplicate => { return Ok(IngestResult { event_id: event.id.to_hex(), @@ -770,6 +769,7 @@ async fn handle_workflow_def( state .db .upsert_workflow( + &mut tx, community_id, workflow_id, Some(channel_id), @@ -777,6 +777,7 @@ async fn handle_workflow_def( &workflow_name, &definition_json_final, &hash, + event.id.as_bytes(), ) .await .map_err(|e| match e { @@ -786,17 +787,15 @@ async fn handle_workflow_def( other => IngestError::Internal(format!("error: db upsert_workflow: {other}")), })?; - // Drop the trigger-path cache entry so the new/updated definition fires on - // the next matching event instead of after the cache TTL. - state - .workflow_engine - .invalidate_channel_workflows(community_id, channel_id); - - // Commit the event transaction after the idempotent workflow upsert succeeds. + // Commit before cache invalidation so a concurrent refill can observe the new revision. tx.commit() .await .map_err(|e| IngestError::Internal(format!("error: commit transaction: {e}")))?; + state + .workflow_engine + .invalidate_channel_workflows(community_id, channel_id); + // 5. Return response let mut resp = serde_json::json!({ "workflow_id": workflow_id.to_string(), @@ -916,6 +915,7 @@ async fn handle_workflow_trigger( .create_workflow_run( community_id, workflow_id, + workflow.definition_event_id.as_deref(), Some(&event_id_bytes), trigger_ctx_json.as_ref(), ) diff --git a/crates/buzz-workflow/src/lib.rs b/crates/buzz-workflow/src/lib.rs index ee1c746776..390b9b14ac 100644 --- a/crates/buzz-workflow/src/lib.rs +++ b/crates/buzz-workflow/src/lib.rs @@ -407,6 +407,7 @@ impl WorkflowEngine { .create_workflow_run( community_id, workflow.id, + workflow.definition_event_id.as_deref(), Some(&trigger_event_id_bytes), Some(&trigger_ctx_json), ) @@ -669,6 +670,7 @@ impl WorkflowEngine { .create_workflow_run( community_id, workflow.id, + workflow.definition_event_id.as_deref(), None, // no trigger event for cron trigger_ctx_json.as_ref(), ) diff --git a/migrations/0040_workflow_revision_binding.sql b/migrations/0040_workflow_revision_binding.sql new file mode 100644 index 0000000000..799e18dae2 --- /dev/null +++ b/migrations/0040_workflow_revision_binding.sql @@ -0,0 +1,8 @@ +-- Preserve exact signed workflow revisions without changing legacy execution. +-- Existing rows remain NULL until an explicit, provenance-safe migration. +ALTER TABLE workflows + ADD COLUMN definition_event_id BYTEA + CHECK (definition_event_id IS NULL OR octet_length(definition_event_id) = 32); +ALTER TABLE workflow_runs + ADD COLUMN definition_event_id BYTEA + CHECK (definition_event_id IS NULL OR octet_length(definition_event_id) = 32); diff --git a/schema/schema.sql b/schema/schema.sql index 0950812562..1c010554b4 100644 --- a/schema/schema.sql +++ b/schema/schema.sql @@ -368,6 +368,9 @@ CREATE TABLE workflows ( channel_id UUID, definition JSONB NOT NULL, definition_hash BYTEA NOT NULL, + definition_event_id BYTEA CHECK ( + definition_event_id IS NULL OR octet_length(definition_event_id) = 32 + ), status workflow_status NOT NULL DEFAULT 'active', enabled BOOLEAN NOT NULL DEFAULT TRUE, created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(), @@ -388,6 +391,9 @@ 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_event_id BYTEA CHECK ( + definition_event_id IS NULL OR octet_length(definition_event_id) = 32 + ), status run_status NOT NULL DEFAULT 'pending', trigger_event_id BYTEA, current_step INT NOT NULL DEFAULT 0,