diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 25a59c32432..8c79231a792 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -697,6 +697,16 @@ jobs: env: DATABASE_URL: postgres://buzz:${{ env.BUZZ_TEST_POSTGRES_PASSWORD }}@localhost:5432/buzz TEST_DATABASE_URL: postgres://buzz:${{ env.BUZZ_TEST_POSTGRES_PASSWORD }}@localhost:5432/buzz + - name: Workflow revision rollout PostgreSQL tests + # Keep mixed-version invalidation and atomic rebinding in the permanent + # Postgres gate; these cases are ignored by infrastructure-free units. + run: | + cargo nextest run \ + --archive-file target/ci/backend-integration-tests.tar.zst \ + -E 'package(buzz-db) and test(/^store::workflow::/)' \ + --run-ignored ignored-only + env: + BUZZ_TEST_DATABASE_URL: postgres://buzz:${{ env.BUZZ_TEST_POSTGRES_PASSWORD }}@localhost:5432/buzz - name: Database pressure observability PostgreSQL tests # Explicit pool acquisition and advisory-lock metrics require real # Postgres and are ignored by the infrastructure-free unit-test job. diff --git a/crates/buzz-db/src/runtime/migration.rs b/crates/buzz-db/src/runtime/migration.rs index 66251563cbd..dcb6a41359e 100644 --- a/crates/buzz-db/src/runtime/migration.rs +++ b/crates/buzz-db/src/runtime/migration.rs @@ -690,7 +690,7 @@ mod tests { let mut migrations: Vec<_> = MIGRATOR.iter().collect(); migrations.sort_by_key(|migration| migration.version); - assert_eq!(migrations.len(), 40); + assert_eq!(migrations.len(), 41); assert_eq!(migrations[0].version, 1); assert_eq!(&*migrations[0].description, "initial schema"); assert!(migrations[0] @@ -1153,6 +1153,35 @@ mod 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[40].version, 41); + let workflow_revision_binding = migrations[40].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.contains("UPDATE workflows")); + let revision_guard = |sql: &str| { + let start = sql + .find("CREATE FUNCTION invalidate_workflow_revision()") + .unwrap(); + let end = sql[start..] + .find("FOR EACH ROW EXECUTE FUNCTION invalidate_workflow_revision();") + .unwrap(); + sql[start..start + end].to_owned() + }; + assert_eq!( + revision_guard(workflow_revision_binding), + revision_guard(desired_schema) + ); + 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 0ae1b623764..6b8dcba1cd0 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() { @@ -356,6 +365,19 @@ pub async fn upsert_workflow( ))); } + // The database invalidates provenance on every materialization UPDATE, + // including old-pod and equal-value writes. Rebind only after that write, + // while this transaction still owns the row lock. Readers cannot see the + // intermediate NULL, and a later legacy writer will clear the binding. + sqlx::query( + "UPDATE workflows SET definition_event_id = $3 WHERE community_id = $1 AND id = $2", + ) + .bind(community_id.as_uuid()) + .bind(id) + .bind(definition_event_id) + .execute(&mut **tx) + .await?; + Ok(()) } @@ -372,7 +394,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 +425,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 +456,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 +484,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 +826,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 +835,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 +859,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 +891,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 +1209,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 +1229,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 +1275,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 +1303,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 +1311,7 @@ impl Db { &self.pool, community_id, workflow_id, + definition_event_id, trigger_event_id, trigger_context, ) @@ -1469,6 +1497,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 +1505,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 +1516,7 @@ impl Db { name, definition_json, definition_hash, + definition_event_id, ) .await } @@ -1685,6 +1716,10 @@ impl Db { // -- Tests -------------------------------------------------------------------- +#[cfg(test)] +#[path = "workflow_revision_tests.rs"] +mod revision_tests; + #[cfg(test)] mod tests { use super::*; @@ -1798,6 +1833,7 @@ mod 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 +1864,7 @@ mod 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 +1887,7 @@ mod 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 +1917,7 @@ mod 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 +1938,7 @@ mod 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 +1962,7 @@ mod 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 +1994,7 @@ mod 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 +2019,7 @@ mod 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 +2052,7 @@ mod 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 +2076,7 @@ mod 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 +2303,85 @@ mod 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 +2504,7 @@ mod 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 +2533,7 @@ mod 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 +2814,10 @@ mod 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-db/src/store/workflow_revision_tests.rs b/crates/buzz-db/src/store/workflow_revision_tests.rs new file mode 100644 index 00000000000..dbec1db2a7d --- /dev/null +++ b/crates/buzz-db/src/store/workflow_revision_tests.rs @@ -0,0 +1,243 @@ +//! Mixed-version writers must never leave an exact revision on legacy materialization. +use super::*; + +async fn pool() -> PgPool { + let url = std::env::var("BUZZ_TEST_DATABASE_URL") + .or_else(|_| std::env::var("DATABASE_URL")) + .expect("test database URL"); + PgPool::connect(&url).await.expect("connect") +} + +async fn fixture(pool: &PgPool) -> WorkflowRecord { + let community = CommunityId::from_uuid(Uuid::new_v4()); + sqlx::query("INSERT INTO communities (id, host) VALUES ($1, $2)") + .bind(community.as_uuid()) + .bind(format!("revision-{}.example", community.as_uuid())) + .execute(pool) + .await + .unwrap(); + let owner = [0xa1; 32]; + crate::user::ensure_user(pool, community, &owner) + .await + .unwrap(); + let id = Uuid::new_v4(); + let mut tx = pool.begin().await.unwrap(); + upsert_workflow( + &mut tx, + community, + id, + None, + &owner, + "revision-test", + r#"{"trigger":{"on":"schedule"},"steps":[]}"#, + &[0; 32], + &[0x42; 32], + ) + .await + .unwrap(); + tx.commit().await.unwrap(); + get_workflow(pool, community, id).await.unwrap() +} + +// The predecessor's actual ON CONFLICT statement: no reference to the new +// column. In particular, do not simulate it by explicitly writing NULL. +async fn legacy_upsert( + tx: &mut sqlx::Transaction<'_, sqlx::Postgres>, + w: &WorkflowRecord, + definition: &str, +) -> Result<()> { + 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) + ON CONFLICT (community_id, id) DO UPDATE + SET name = EXCLUDED.name, + definition = EXCLUDED.definition, + definition_hash = EXCLUDED.definition_hash, + updated_at = NOW() + WHERE workflows.owner_pubkey = EXCLUDED.owner_pubkey + AND workflows.channel_id IS NOT DISTINCT FROM EXCLUDED.channel_id + RETURNING id + "#) + .bind(w.community_id.as_uuid()).bind(w.id).bind(&w.name) + .bind(&w.owner_pubkey).bind(w.channel_id).bind(definition) + .bind(&w.definition_hash).fetch_one(&mut **tx).await?; + Ok(()) +} + +async fn rebind(pool: &PgPool, w: &WorkflowRecord) { + let mut tx = pool.begin().await.unwrap(); + upsert_workflow( + &mut tx, + w.community_id, + w.id, + w.channel_id, + &w.owner_pubkey, + &w.name, + &w.definition.to_string(), + &w.definition_hash, + &[0x43; 32], + ) + .await + .unwrap(); + tx.commit().await.unwrap(); +} + +#[tokio::test] +#[ignore = "requires Postgres"] +async fn legacy_materialization_clears_revision_even_when_values_match() { + let pool = pool().await; + let w = fixture(&pool).await; + assert_eq!(w.definition_event_id, Some(vec![0x42; 32])); + let original_run = create_workflow_run( + &pool, + w.community_id, + w.id, + w.definition_event_id.as_deref(), + None, + None, + ) + .await + .unwrap(); + + for definition in [ + w.definition.to_string(), + r#"{"steps":[{"id":"different"}]}"#.to_owned(), + ] { + rebind(&pool, &w).await; + assert_eq!( + get_workflow(&pool, w.community_id, w.id) + .await + .unwrap() + .definition_event_id, + Some(vec![0x43; 32]) + ); + let mut tx = pool.begin().await.unwrap(); + legacy_upsert(&mut tx, &w, &definition).await.unwrap(); + tx.commit().await.unwrap(); + let current = get_workflow(&pool, w.community_id, w.id).await.unwrap(); + assert!(current.definition_event_id.is_none()); + assert_eq!( + current.definition, + serde_json::from_str::(&definition).unwrap() + ); + let run = create_workflow_run( + &pool, + w.community_id, + w.id, + current.definition_event_id.as_deref(), + None, + None, + ) + .await + .unwrap(); + assert!(get_workflow_run(&pool, w.community_id, run) + .await + .unwrap() + .definition_event_id + .is_none()); + } + // Already-created runs keep the revision they actually selected. + assert_eq!( + get_workflow_run(&pool, w.community_id, original_run) + .await + .unwrap() + .definition_event_id, + Some(vec![0x42; 32]) + ); + + rebind(&pool, &w).await; + sqlx::query("UPDATE workflows SET enabled = FALSE, status = 'disabled' WHERE community_id = $1 AND id = $2") + .bind(w.community_id.as_uuid()).bind(w.id).execute(&pool).await.unwrap(); + assert_eq!( + get_workflow(&pool, w.community_id, w.id) + .await + .unwrap() + .definition_event_id, + Some(vec![0x43; 32]) + ); + // The sibling materialization helper must also invalidate, not just upsert. + update_workflow( + &pool, + w.community_id, + w.id, + &w.name, + &w.definition.to_string(), + &w.definition_hash, + ) + .await + .unwrap(); + assert!(get_workflow(&pool, w.community_id, w.id) + .await + .unwrap() + .definition_event_id + .is_none()); +} + +#[tokio::test] +#[ignore = "requires Postgres"] +async fn revision_rebind_is_atomic_against_legacy_writer_and_rollback() { + let pool = pool().await; + let w = fixture(&pool).await; + let mut new_tx = pool.begin().await.unwrap(); + upsert_workflow( + &mut new_tx, + w.community_id, + w.id, + w.channel_id, + &w.owner_pubkey, + &w.name, + r#"{"steps":[{"id":"new"}]}"#, + &[1; 32], + &[0x43; 32], + ) + .await + .unwrap(); + // A separate reader sees the complete old row, never the intermediate NULL. + let visible = get_workflow(&pool, w.community_id, w.id).await.unwrap(); + assert_eq!(visible.definition, w.definition); + assert_eq!(visible.definition_event_id, w.definition_event_id); + let mut old_tx = pool.begin().await.unwrap(); + sqlx::query("SET LOCAL lock_timeout = '100ms'") + .execute(&mut *old_tx) + .await + .unwrap(); + let error = legacy_upsert(&mut old_tx, &w, &w.definition.to_string()) + .await + .unwrap_err(); + assert!(error.to_string().contains("lock timeout"), "{error}"); + old_tx.rollback().await.unwrap(); + new_tx.rollback().await.unwrap(); + assert_eq!( + get_workflow(&pool, w.community_id, w.id) + .await + .unwrap() + .definition_event_id, + w.definition_event_id + ); + + rebind(&pool, &w).await; + let mut old_tx = pool.begin().await.unwrap(); + legacy_upsert(&mut old_tx, &w, &w.definition.to_string()) + .await + .unwrap(); + // Rollback of an old writer restores the previous binding as well. + old_tx.rollback().await.unwrap(); + assert_eq!( + get_workflow(&pool, w.community_id, w.id) + .await + .unwrap() + .definition_event_id, + Some(vec![0x43; 32]) + ); + let mut old_tx = pool.begin().await.unwrap(); + legacy_upsert(&mut old_tx, &w, &w.definition.to_string()) + .await + .unwrap(); + old_tx.commit().await.unwrap(); + assert!(get_workflow(&pool, w.community_id, w.id) + .await + .unwrap() + .definition_event_id + .is_none()); +} diff --git a/crates/buzz-relay/src/api/bridge.rs b/crates/buzz-relay/src/api/bridge.rs index 5dbb2aaf50c..57b65320758 100644 --- a/crates/buzz-relay/src/api/bridge.rs +++ b/crates/buzz-relay/src/api/bridge.rs @@ -2115,7 +2115,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 ae7adc98143..5ce845af5b4 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 bceb6d8bd8d..4fc4f9d5be8 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/0041_workflow_revision_binding.sql b/migrations/0041_workflow_revision_binding.sql new file mode 100644 index 00000000000..20582970019 --- /dev/null +++ b/migrations/0041_workflow_revision_binding.sql @@ -0,0 +1,24 @@ +-- Preserve exact signed workflow revisions without changing legacy execution. +-- Existing rows remain NULL until a new signed definition is ingested. +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); + +-- A legacy writer does not mention definition_event_id. Column-targeted triggers +-- fire even for equal-value rewrites, where comparing OLD/NEW would invent +-- provenance. New writers rebind separately while still holding the row lock +-- in the signed-event transaction. Operational status/enabled updates keep it. +CREATE FUNCTION invalidate_workflow_revision() RETURNS TRIGGER AS $$ +BEGIN + NEW.definition_event_id := NULL; + RETURN NEW; +END; +$$ LANGUAGE plpgsql; + +CREATE TRIGGER workflows_invalidate_revision +BEFORE UPDATE OF id, community_id, owner_pubkey, channel_id, name, definition, definition_hash +ON workflows +FOR EACH ROW EXECUTE FUNCTION invalidate_workflow_revision(); diff --git a/schema/schema.sql b/schema/schema.sql index 54566103335..3e876661e75 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(), @@ -377,6 +380,23 @@ CREATE TABLE workflows ( FOREIGN KEY (community_id, channel_id) REFERENCES channels (community_id, id) ); + +-- A legacy writer does not mention definition_event_id. Column-targeted triggers +-- fire even for equal-value rewrites, where comparing OLD/NEW would invent +-- provenance. New writers rebind separately while still holding the row lock +-- in the signed-event transaction. Operational status/enabled updates keep it. +CREATE FUNCTION invalidate_workflow_revision() RETURNS TRIGGER AS $$ +BEGIN + NEW.definition_event_id := NULL; + RETURN NEW; +END; +$$ LANGUAGE plpgsql; + +CREATE TRIGGER workflows_invalidate_revision +BEFORE UPDATE OF id, community_id, owner_pubkey, channel_id, name, definition, definition_hash +ON workflows +FOR EACH ROW EXECUTE FUNCTION invalidate_workflow_revision(); + CREATE INDEX idx_workflows_channel_active ON workflows (community_id, channel_id, status, enabled); -- Scheduler scans enabled schedule workflows; community_id returned per row so -- side effects run under the owning tenant's context (Lane0 contract §4a.5). @@ -388,6 +408,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,