jerry/buzz
1
0
Fork 0
mirror of https://github.com/block/buzz.git synced 2026-11-04 00:12:02 +00:00
Code Issues Packages Projects Releases Wiki Activity

Persist exact workflow revisions additively

Browse Source 
Signed-off-by: Logan Johnson <loganj@squareup.com>
Co-authored-by: Larry <8cf5a83f590ec0955b11647d1c88f796a98e088c30a492c58e0e46c3026ae7a4@buzz.block.builderlab.xyz>
Signed-off-by: Larry <8cf5a83f590ec0955b11647d1c88f796a98e088c30a492c58e0e46c3026ae7a4@buzz.block.builderlab.xyz>
Signed-off-by: Logan Johnson <loganj@squareup.com>
This commit is contained in:
Logan Johnson
2026-08-27 17:07:36 -04:00
parent b1f6b7ef77
commit 72f92568e8
7 changed files with 183 additions and 34 deletions
Download Patch File Download Diff File Expand all files Collapse all files

18
crates/buzz-db/src/runtime/migration.rs
View File

@@ -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.

145
crates/buzz-db/src/store/workflow.rs
View File

@@ -179,6 +179,9 @@ pub struct WorkflowRecord {
pub definition: serde_json::Value,
/// SHA-256 hash of the canonical definition JSON.
pub definition_hash: Vec<u8>,
/// Exact owner-signed kind:30620 event that materialized this revision.
/// NULL is retained for workflows created before revision capture.
pub definition_event_id: Option<Vec<u8>>,
/// 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<Vec<u8>>,
/// 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<Uuid>,
@@ -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<WorkflowRecord> {
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<Vec<WorkflowRecord>> {
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<Vec<WorkflowRecord>> {
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<Uuid> {
@@ -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<WorkflowRunRecord> {
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<WorkflowRecord>
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<WorkflowRunRecord> {
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<Option<WorkflowRecord>> {
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<Uuid> {
@@ -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<Uuid>,
@@ -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");

8
crates/buzz-relay/src/api/bridge.rs
View File

@@ -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}")))?;

30
crates/buzz-relay/src/handlers/command_executor.rs
View File

@@ -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(),
)

2
crates/buzz-workflow/src/lib.rs
View File

@@ -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(),
)

8
migrations/0040_workflow_revision_binding.sql Normal file
View File

@@ -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);

6
schema/schema.sql
View File

@@ -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,