From b064beb5c39a8ea52c1d43565ddcace10f780ed6 Mon Sep 17 00:00:00 2001 From: Logan Johnson Date: Fri, 28 Aug 2026 10:15:59 -0400 Subject: [PATCH] Fence legacy workflow revision reconciliation Signed-off-by: Logan Johnson Co-authored-by: Larry <8cf5a83f590ec0955b11647d1c88f796a98e088c30a492c58e0e46c3026ae7a4@buzz.block.builderlab.xyz> --- crates/buzz-db/src/workflow.rs | 41 ++++++++++++++++++- .../src/handlers/command_executor.rs | 28 +++++++------ 2 files changed, 55 insertions(+), 14 deletions(-) diff --git a/crates/buzz-db/src/workflow.rs b/crates/buzz-db/src/workflow.rs index 1494a48dc6..39c7f7303a 100644 --- a/crates/buzz-db/src/workflow.rs +++ b/crates/buzz-db/src/workflow.rs @@ -424,13 +424,14 @@ pub async fn bind_legacy_workflow_revision( let result = sqlx::query( r#" UPDATE workflows - SET definition_event_id = $7 + SET definition_event_id = $8 WHERE community_id = $1 AND id = $2 AND owner_pubkey = $3 AND channel_id IS NOT DISTINCT FROM $4 AND definition = $5 AND definition_hash = $6 + AND updated_at = $7 AND definition_event_id IS NULL "#, ) @@ -440,6 +441,7 @@ pub async fn bind_legacy_workflow_revision( .bind(workflow.channel_id) .bind(&workflow.definition) .bind(&workflow.definition_hash) + .bind(workflow.updated_at) .bind(definition_event_id) .execute(pool) .await?; @@ -1997,6 +1999,43 @@ mod tests { assert_eq!(current.definition_hash, rewritten_hash); } + #[tokio::test] + #[ignore = "requires Postgres"] + async fn legacy_revision_binding_rejects_an_equal_definition_rewrite() { + let pool = setup_pool().await; + let community = make_community(&pool).await; + let (workflow_id, _) = make_workflow_in(&pool, community).await; + let snapshot = get_workflow(&pool, community, workflow_id) + .await + .expect("read legacy workflow snapshot"); + + sqlx::query( + r#" + UPDATE workflows + SET definition = definition, + definition_hash = definition_hash, + updated_at = updated_at + INTERVAL '1 microsecond' + WHERE community_id = $1 AND id = $2 + "#, + ) + .bind(community.as_uuid()) + .bind(workflow_id) + .execute(&pool) + .await + .expect("simulate equal-materialization legacy rewrite"); + + assert!( + !bind_legacy_workflow_revision(&pool, &snapshot, &[0x42; 32]) + .await + .expect("reject stale generation") + ); + assert!(get_workflow(&pool, community, workflow_id) + .await + .expect("read rewritten workflow") + .definition_event_id + .is_none()); + } + #[tokio::test] #[ignore = "requires Postgres"] async fn revision_capture_is_additive_and_transactional() { diff --git a/crates/buzz-relay/src/handlers/command_executor.rs b/crates/buzz-relay/src/handlers/command_executor.rs index 18b74853f0..ed92fd07fd 100644 --- a/crates/buzz-relay/src/handlers/command_executor.rs +++ b/crates/buzz-relay/src/handlers/command_executor.rs @@ -328,16 +328,12 @@ fn legacy_definition_matches(workflow: &WorkflowRecord, event: &Event) -> bool { // Webhook secrets were generated by the relay and never existed in the // owner-signed event. Reuse the stored secret before comparing the exact // materialized JSON and its legacy hash. - let materialized_json = - if let Some(secret) = webhook_secret::extract_secret(&workflow.definition) { - webhook_secret::inject_secret(&mut definition, &secret); - let Ok(json) = serde_json::to_string(&definition) else { - return false; - }; - json - } else { - canonical_json - }; + if let Some(secret) = webhook_secret::extract_secret(&workflow.definition) { + webhook_secret::inject_secret(&mut definition, &secret); + } + let Ok(materialized_json) = serde_json::to_string(&definition) else { + return false; + }; definition == workflow.definition && compute_definition_hash(&materialized_json) == workflow.definition_hash @@ -1532,6 +1528,8 @@ mod tests { .expect("canonical workflow definition"); let definition: serde_json::Value = serde_json::from_str(&canonical_json).expect("definition json"); + let materialized_json = + serde_json::to_string(&definition).expect("production materialized json"); WorkflowRecord { id: Uuid::parse_str(extract_d_tag(event).as_deref().expect("d tag")) .expect("workflow UUID"), @@ -1539,7 +1537,7 @@ mod tests { name: "legacy".to_string(), owner_pubkey: event.pubkey.to_bytes().to_vec(), channel_id: Some(channel_id), - definition_hash: compute_definition_hash(&canonical_json), + definition_hash: compute_definition_hash(&materialized_json), definition, definition_event_id: None, status: buzz_db::workflow::WorkflowStatus::Active, @@ -1626,14 +1624,18 @@ mod tests { let content = "name: legacy\ntrigger:\n on: message_posted\nsteps:\n - id: notify\n action: send_message\n text: hello\n"; let (_, canonical_json) = buzz_workflow::WorkflowEngine::parse_yaml(content) .expect("canonical workflow definition"); + let definition: serde_json::Value = + serde_json::from_str(&canonical_json).expect("production definition value"); + let definition_json_final = + serde_json::to_string(&definition).expect("production materialized definition"); let workflow_id = db .create_workflow( tenant.community(), Some(channel.id), owner.as_slice(), "legacy", - &canonical_json, - &compute_definition_hash(&canonical_json), + &definition_json_final, + &compute_definition_hash(&definition_json_final), ) .await .expect("create legacy workflow");