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

Keep approval resumes on signed workflow revisions

Browse Source 
Co-authored-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 21:30:57 -04:00
parent 46b2f7cb5d
commit d8d7004093
2 changed files with 264 additions and 24 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -1439,19 +1439,12 @@ async fn resume_workflow_after_approval(
return;
}
let workflow = match db.get_workflow(community_id, workflow_id).await {
Ok(w) => w,
let (run, def) = match engine.load_run_definition(community_id, run_id).await {
Ok(loaded) => loaded,
Err(e) => {
tracing::error!("resume_workflow: failed to fetch workflow {workflow_id}: {e}");
return;
}
};
let def: buzz_workflow::WorkflowDef = match serde_json::from_value(workflow.definition.clone())
{
Ok(d) => d,
Err(e) => {
tracing::error!("resume_workflow: failed to parse workflow definition: {e}");
tracing::error!(
"resume_workflow: failed to load signed definition for run {run_id}: {e}"
);
if let Err(db_err) = db
.update_workflow_run(
community_id,
@@ -1461,7 +1454,7 @@ async fn resume_workflow_after_approval(
&run.execution_trace,
Some(buzz_db::workflow::WorkflowRunFailure {
code: "invalid_definition",
message: &format!("definition parse error: {e}"),
message: &format!("signed run definition unavailable: {e}"),
}),
)
.await
@@ -1472,6 +1465,30 @@ async fn resume_workflow_after_approval(
}
};
if run.workflow_id != workflow_id {
tracing::error!(
"resume_workflow: approval workflow {workflow_id} does not match run workflow {}",
run.workflow_id
);
if let Err(e) = db
.update_workflow_run(
community_id,
run_id,
RunStatus::Failed,
run.current_step,
&run.execution_trace,
Some(buzz_db::workflow::WorkflowRunFailure {
code: "approval_binding_mismatch",
message: "approval does not belong to the workflow run",
}),
)
.await
{
tracing::error!("resume_workflow: failed to mark mismatched run as failed: {e}");
}
return;
}
// Reconstruct step_outputs from execution trace for template resolution
let mut initial_outputs: std::collections::HashMap<String, serde_json::Value> =
std::collections::HashMap::new();
@@ -1515,6 +1532,34 @@ mod postgres_tests {
use super::*;
use nostr::{EventBuilder, Keys, Kind, Tag, Timestamp};
#[derive(Default)]
struct RecordingActionSink {
messages: std::sync::Mutex<Vec<String>>,
}
impl buzz_workflow::ActionSink for RecordingActionSink {
fn send_message(
&self,
_community_id: CommunityId,
_channel_id: &str,
text: &str,
_author_pubkey: &str,
_reply_to: Option<&str>,
) -> std::pin::Pin<
Box<
dyn std::future::Future<Output = Result<String, buzz_workflow::ActionSinkError>>
+ Send
+ '_,
>,
> {
self.messages
.lock()
.expect("recording action sink lock")
.push(text.to_string());
Box::pin(async { Ok("recorded-event".to_string()) })
}
}
async fn persistence_test_context() -> (buzz_db::Db, TenantContext) {
let url = std::env::var("BUZZ_TEST_DATABASE_URL")
.or_else(|_| std::env::var("DATABASE_URL"))
@@ -1597,7 +1642,9 @@ mod postgres_tests {
concat!(
"name: manual-trigger-pool\n",
"trigger:\n on: message_posted\n",
"steps:\n - id: send\n action: send_message\n text: done\n",
"steps:\n",
" - id: approval\n action: request_approval\n from: '@owner'\n message: approve\n",
" - id: send\n action: send_message\n text: done\n",
),
)
.tags(vec![
@@ -1790,6 +1837,163 @@ mod postgres_tests {
assert!(agent_result.message.contains("\"run_id\""));
}
#[tokio::test]
#[ignore = "requires Postgres"]
async fn approval_resume_executes_the_run_bound_signed_revision() {
let (state, tenant, _human, agent, workflow_id, revision_a) =
manual_trigger_test_context().await;
let community_id = tenant.community();
let db = state.db.clone();
let trigger_context = serde_json::to_value(TriggerContext {
channel_id: exact_tag_value(&revision_a, "h")
.unwrap_or_default()
.to_string(),
..TriggerContext::default()
})
.expect("serialize trigger context");
let run_id = db
.create_workflow_run(
community_id,
workflow_id,
Some(revision_a.id.as_bytes()),
None,
Some(&trigger_context),
)
.await
.expect("create revision A run");
db.update_workflow_run(
community_id,
run_id,
RunStatus::WaitingApproval,
0,
&serde_json::json!([{
"step_id": "approval",
"output": {"approved": true}
}]),
None,
)
.await
.expect("suspend revision A run for approval");
let channel_id = Uuid::parse_str(exact_tag_value(&revision_a, "h").expect("channel tag"))
.expect("channel UUID");
let revision_b = EventBuilder::new(
Kind::Custom(KIND_WORKFLOW_DEF as u16),
concat!(
"name: revision-b\n",
"trigger:\n on: message_posted\n",
"steps:\n",
" - id: approval\n action: request_approval\n from: '@owner'\n message: approve\n",
" - id: after\n action: send_message\n text: revision B\n",
),
)
.tags(vec![
Tag::parse(["d", workflow_id.to_string().as_str()]).expect("d tag"),
Tag::parse(["h", channel_id.to_string().as_str()]).expect("h tag"),
])
.sign_with_keys(&agent)
.expect("sign revision B");
let (_, definition_b_json) = buzz_workflow::WorkflowEngine::parse_yaml(&revision_b.content)
.expect("parse revision B");
let definition_b_hash = compute_definition_hash(&definition_b_json);
let mut tx = db
.begin_transaction()
.await
.expect("begin revision B update");
buzz_db::event::insert_event_in_transaction(
&mut tx,
community_id,
&revision_b,
Some(channel_id),
)
.await
.expect("persist revision B");
db.upsert_workflow(
&mut tx,
community_id,
workflow_id,
Some(channel_id),
&agent.public_key().to_bytes(),
"revision-b",
&definition_b_json,
&definition_b_hash,
revision_b.id.as_bytes(),
)
.await
.expect("materialize revision B");
tx.commit().await.expect("commit revision B update");
let sink = Arc::new(RecordingActionSink::default());
state.workflow_engine.set_action_sink(sink.clone());
resume_workflow_after_approval(
Arc::clone(&state.workflow_engine),
db.clone(),
community_id,
run_id,
workflow_id,
1,
)
.await;
let resumed = db
.get_workflow_run(community_id, run_id)
.await
.expect("load resumed run");
assert_eq!(resumed.status, RunStatus::Completed);
assert_eq!(
sink.messages
.lock()
.expect("recorded messages lock")
.as_slice(),
["done"],
"approval resume must execute revision A, never current revision B"
);
}
#[tokio::test]
#[ignore = "requires Postgres"]
async fn approval_resume_fails_closed_without_a_signed_run_revision() {
let (state, tenant, _human, _agent, workflow_id, _revision) =
manual_trigger_test_context().await;
let community_id = tenant.community();
let db = state.db.clone();
let run_id = db
.create_workflow_run(community_id, workflow_id, None, None, None)
.await
.expect("create legacy revisionless run");
db.update_workflow_run(
community_id,
run_id,
RunStatus::WaitingApproval,
0,
&serde_json::json!([]),
None,
)
.await
.expect("suspend revisionless run");
resume_workflow_after_approval(
Arc::clone(&state.workflow_engine),
db.clone(),
community_id,
run_id,
workflow_id,
1,
)
.await;
let failed = db
.get_workflow_run(community_id, run_id)
.await
.expect("load failed run");
assert_eq!(failed.status, RunStatus::Failed);
assert_eq!(failed.error_code.as_deref(), Some("invalid_definition"));
assert!(failed
.error_message
.as_deref()
.is_some_and(|message| message.contains("no owner-signed definition revision")));
}
#[tokio::test]
#[ignore = "requires Postgres"]
async fn manual_trigger_rejects_non_owner_and_stale_or_missing_revision() {

56
crates/buzz-test-client/tests/conformance_multitenant.rs
View File

@@ -1778,7 +1778,31 @@ mod workflows {
.to_string()
}
/// Fire a workflow by id on `http_base`'s community (kind:46020, `d`=id).
/// Fetch the current owner-signed workflow revision from `http_base`'s
/// community. Manual triggers must name this exact event id.
async fn workflow_revision(http_base: &str, keys: &Keys, workflow_id: &str) -> String {
let resp = reqwest::Client::new()
.get(format!("{http_base}/workflows/{workflow_id}/revision"))
.header("X-Pubkey", keys.public_key().to_hex())
.send()
.await
.unwrap_or_else(|e| panic!("GET workflow revision from {http_base} failed: {e}"));
let status = resp.status();
let body = resp.text().await.expect("read workflow revision body");
assert!(
status.is_success(),
"workflow revision lookup against {http_base} failed with {status}: {body}"
);
let event: serde_json::Value = serde_json::from_str(&body)
.unwrap_or_else(|e| panic!("parse workflow revision JSON: {e} ({body:?})"));
event["id"]
.as_str()
.unwrap_or_else(|| panic!("workflow revision response missing event id: {event}"))
.to_string()
}
/// Fire an exact workflow revision on `http_base`'s community
/// (kind:46020, `d`=workflow id, `e`=signed revision id).
/// Returns a normalized `{accepted, message}` body so the caller can assert
/// on the *wire-observable* accept/reject and message. The HTTP bridge maps
/// `IngestError::Rejected` to HTTP 400 + `{error}` while the WS door maps the
@@ -1788,9 +1812,13 @@ mod workflows {
http_base: &str,
keys: &Keys,
workflow_id: &str,
revision: &str,
) -> serde_json::Value {
let event = EventBuilder::new(Kind::Custom(KIND_WORKFLOW_TRIGGER), "")
.tags(vec![Tag::parse(["d", workflow_id]).unwrap()])
.tags(vec![
Tag::parse(["d", workflow_id]).unwrap(),
Tag::parse(["e", revision]).unwrap(),
])
.sign_with_keys(keys)
.unwrap();
@@ -1837,9 +1865,11 @@ mod workflows {
/// This deliberately removes "not a member of U in B" as an alternate
/// cause of the B rejection — K *is* a member of U in B.
/// 2. Define a workflow in `U` under **A** (kind:30620). The server
/// generates `W` and returns it. `W` is an A-community row.
/// 3. Fire `W` under host **B** (kind:46020, `d`=W) as K. Must be
/// rejected — `accepted == false` and the generic `workflow not found`
/// generates `W` and returns it. Retrieve its current signed revision
/// `R` through A's authenticated revision endpoint. `W` and `R` are
/// A-community records.
/// 3. Fire `W`/`R` under host **B** (kind:46020, `d`=W, `e`=R) as K. Must
/// be rejected — `accepted == false` and the generic `workflow not found`
/// message — because `get_workflow(B_community, W)` finds nothing: `W`
/// exists only in A. K's membership of U-in-B is irrelevant; the
/// lookup never reaches the membership check.
@@ -1893,11 +1923,17 @@ mod workflows {
uuid::Uuid::parse_str(&workflow_id).is_ok(),
"server-generated workflow_id must be a UUID, got {workflow_id:?}"
);
let revision = workflow_revision(&http_a, &keys, &workflow_id).await;
assert_eq!(
revision.len(),
64,
"signed workflow revision must be a 32-byte event id"
);
// (3) Fire W under host B as K. Must fail closed: W is an A-community
// row, and get_workflow(B_community, W) finds nothing. K is a member of
// U in B, so a leak here is the community fence failing, not membership.
let b_resp = trigger_workflow(&http_b, &keys, &workflow_id).await;
// (3) Fire W/R under host B as K. Keep the trigger grammar otherwise
// valid with A's real signed revision so only tenant-scoped workflow
// resolution can explain the rejection.
let b_resp = trigger_workflow(&http_b, &keys, &workflow_id, &revision).await;
assert_eq!(
b_resp["accepted"].as_bool(),
Some(false),
@@ -1915,7 +1951,7 @@ mod workflows {
// proves the B rejection is community confinement, not an
// untriggerable workflow, and exercises the same-community happy path
// through the fence under test.
let a_resp = trigger_workflow(&http_a, &keys, &workflow_id).await;
let a_resp = trigger_workflow(&http_a, &keys, &workflow_id, &revision).await;
assert_eq!(
a_resp["accepted"].as_bool(),
Some(true),