From a11e396c2cd935325fdceb1205ca4cf28db8a3b7 Mon Sep 17 00:00:00 2001 From: Logan Johnson Date: Fri, 28 Aug 2026 12:50:57 -0400 Subject: [PATCH] Invalidate workflow revision bindings across legacy writes Clear bindings at the database materialization boundary, even on equal-value legacy updates. Rebind inside atomic signed ingest while holding the row lock. Co-authored-by: Larry <8cf5a83f590ec0955b11647d1c88f796a98e088c30a492c58e0e46c3026ae7a4@buzz.block.builderlab.xyz> Signed-off-by: Logan Johnson Signed-off-by: Larry <8cf5a83f590ec0955b11647d1c88f796a98e088c30a492c58e0e46c3026ae7a4@buzz.block.builderlab.xyz> Signed-off-by: Logan Johnson --- crates/buzz-db/src/runtime/migration.rs | 17 +- crates/buzz-db/src/store/workflow.rs | 17 ++ crates/buzz-db/src/workflow_revision_tests.rs | 243 ++++++++++++++++++ migrations/0040_workflow_revision_binding.sql | 18 +- schema/schema.sql | 17 ++ 5 files changed, 308 insertions(+), 4 deletions(-) create mode 100644 crates/buzz-db/src/workflow_revision_tests.rs diff --git a/crates/buzz-db/src/runtime/migration.rs b/crates/buzz-db/src/runtime/migration.rs index baad0b3e30..bd94abdc60 100644 --- a/crates/buzz-db/src/runtime/migration.rs +++ b/crates/buzz-db/src/runtime/migration.rs @@ -1173,9 +1173,20 @@ mod postgres_tests { 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!(!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") diff --git a/crates/buzz-db/src/store/workflow.rs b/crates/buzz-db/src/store/workflow.rs index 427e5b211b..2fe64aed65 100644 --- a/crates/buzz-db/src/store/workflow.rs +++ b/crates/buzz-db/src/store/workflow.rs @@ -365,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(()) } @@ -1705,6 +1718,10 @@ impl Db { // -- Tests -------------------------------------------------------------------- +#[cfg(test)] +#[path = "workflow_revision_tests.rs"] +mod revision_postgres_tests; + #[cfg(test)] mod postgres_tests { use super::*; diff --git a/crates/buzz-db/src/workflow_revision_tests.rs b/crates/buzz-db/src/workflow_revision_tests.rs new file mode 100644 index 0000000000..dbec1db2a7 --- /dev/null +++ b/crates/buzz-db/src/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/migrations/0040_workflow_revision_binding.sql b/migrations/0040_workflow_revision_binding.sql index 799e18dae2..2058297001 100644 --- a/migrations/0040_workflow_revision_binding.sql +++ b/migrations/0040_workflow_revision_binding.sql @@ -1,8 +1,24 @@ -- Preserve exact signed workflow revisions without changing legacy execution. --- Existing rows remain NULL until an explicit, provenance-safe migration. +-- 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 1c010554b4..67d5d6730b 100644 --- a/schema/schema.sql +++ b/schema/schema.sql @@ -380,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).