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

fix(workflows): atomically fence run admission to selected revision

Browse Source 
Signed-off-by: Pinky <5f5ab050ec58ae208332edd544ebf705221e24c1b86d82a6ca07038a7a8f6ac9@buzz.block.builderlab.xyz>
Signed-off-by: Brain <1a02c72794dcd0f07058a353bc3a81f4028b8c77c92c87fce6d5c8b85970a20b@buzz.block.builderlab.xyz>
This commit is contained in:
Pinky
2026-09-13 12:06:44 -06:00
committed by Brain
parent e604aa4669
commit c73db439d0
7 changed files with 567 additions and 76 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -10,6 +10,9 @@
/// Transactional workflow-coordinate lifecycle operations.
pub mod lifecycle;
mod run_admission;
pub use run_admission::create_workflow_run;
use std::fmt;
use std::str::FromStr;
@@ -867,38 +870,6 @@ pub async fn delete_workflow_for_owner(
// -- Workflow Run CRUD --------------------------------------------------------
/// Insert a new workflow run. Returns the new run's UUID.
///
/// `trigger_context` is the serialized `TriggerContext` for this run. It is stored
/// so that post-approval resume steps can restore the original trigger data and
/// correctly resolve `{{trigger.*}}` template variables.
pub async fn create_workflow_run(
pool: &PgPool,
community_id: CommunityId,
workflow_id: Uuid,
trigger_event_id: Option<&[u8]>,
trigger_context: Option<&serde_json::Value>,
) -> Result<Uuid> {
let id = Uuid::new_v4();
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)
"#,
)
.bind(community_id.as_uuid())
.bind(id)
.bind(workflow_id)
.bind(trigger_event_id)
.bind(trigger_context)
.execute(pool)
.await?;
Ok(id)
}
/// Fetch a single workflow run by ID, scoped to its community.
pub async fn get_workflow_run(
pool: &PgPool,
@@ -1343,19 +1314,18 @@ pub async fn find_by_owner_and_name(
// -- Run and approval Db API --------------------------------------------------
impl Db {
/// Create a new workflow run.
/// Admit a run only for the still-current, active/enabled selected workflow.
/// See [`create_workflow_run`] for the atomic admission contract.
#[datastore_span(name = "create_workflow_run", system = "postgresql")]
pub async fn create_workflow_run(
&self,
community_id: CommunityId,
workflow_id: Uuid,
workflow: &WorkflowRecord,
trigger_event_id: Option<&[u8]>,
trigger_context: Option<&serde_json::Value>,
) -> Result<Uuid> {
) -> Result<Option<Uuid>> {
crate::workflow::create_workflow_run(
&self.pool,
community_id,
workflow_id,
workflow,
trigger_event_id,
trigger_context,
)
@@ -2452,9 +2422,11 @@ 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 workflow = get_workflow(&pool, community, workflow_id).await.unwrap();
let run_id = create_workflow_run(&pool, &workflow, None, None)
.await
.expect("create run ok");
.expect("create run ok")
.expect("current workflow admitted");
let attached =
attach_scheduled_workflow_run(&pool, community, workflow_id, scheduled_for, run_id)
@@ -2481,9 +2453,10 @@ 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, &workflow, None, None)
.await
.expect("create second run ok");
.expect("create second run ok")
.expect("current workflow admitted");
let reattached =
attach_scheduled_workflow_run(&pool, community, workflow_id, scheduled_for, other_run)
.await
@@ -2762,12 +2735,16 @@ 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 workflow_a = get_workflow(&pool, community_a, workflow_id).await.unwrap();
let workflow_b = get_workflow(&pool, community_b, workflow_id).await.unwrap();
let run_a = create_workflow_run(&pool, &workflow_a, None, None)
.await
.expect("run A");
let run_b = create_workflow_run(&pool, community_b, workflow_id, None, None)
.expect("run A")
.expect("current workflow admitted");
let run_b = create_workflow_run(&pool, &workflow_b, None, None)
.await
.expect("run B");
.expect("run B")
.expect("current workflow admitted");
let token = "shared-approval-token";
let expires = Utc::now() + chrono::Duration::hours(1);

251
crates/buzz-db/src/store/workflow/run_admission.rs Normal file
View File

@@ -0,0 +1,251 @@
//! Atomic admission of new work against the selected workflow incarnation.
use sqlx::PgPool;
use uuid::Uuid;
use super::WorkflowRecord;
use crate::Result;
/// Insert a pending run only if the selected workflow is still current and eligible.
///
/// The record must be the one used to select/authorize the execution definition;
/// its community is server-resolved provenance, never a client-supplied tenant.
/// Returns `None` if the workflow is missing, inactive, disabled, or no longer
/// matches the selected definition hash, owner, channel or creation incarnation.
/// Database failures remain errors. Callers must not execute without `Some(id)`.
///
/// A SHARE row lock conflicts with updates and deletion until the INSERT commits
/// (unlike the FK's KEY SHARE lock). PostgreSQL rechecks the predicates after a
/// concurrent writer settles; selection and insertion have no unlocked gap.
/// Already admitted runs are intentionally unaffected by this boundary.
///
/// `trigger_context` stores the serialized `TriggerContext` so approval resume
/// can restore the original trigger data and `{{trigger.*}}` template variables.
pub async fn create_workflow_run(
pool: &PgPool,
workflow: &WorkflowRecord,
trigger_event_id: Option<&[u8]>,
trigger_context: Option<&serde_json::Value>,
) -> Result<Option<Uuid>> {
Ok(sqlx::query_scalar(
r#"
WITH eligible AS (
SELECT community_id, id FROM workflows
WHERE community_id = $1 AND id = $2
AND status = 'active' AND enabled = TRUE
AND definition_hash = $3 AND owner_pubkey = $4
AND channel_id IS NOT DISTINCT FROM $5 AND created_at = $6
FOR SHARE
)
INSERT INTO workflow_runs
(community_id, id, workflow_id, status, trigger_event_id, current_step, execution_trace, trigger_context)
SELECT community_id, $7, id, 'pending', $8, 0, '[]', $9 FROM eligible
RETURNING id
"#,
)
.bind(workflow.community_id.as_uuid())
.bind(workflow.id)
.bind(&workflow.definition_hash)
.bind(&workflow.owner_pubkey)
.bind(workflow.channel_id)
.bind(workflow.created_at)
.bind(Uuid::new_v4())
.bind(trigger_event_id)
.bind(trigger_context)
.fetch_optional(pool)
.await?)
}
#[cfg(test)]
mod postgres_tests {
use super::*;
use crate::{workflow, CommunityId};
use std::time::Duration;
async fn fixture() -> (PgPool, WorkflowRecord) {
let pool = PgPool::connect(&crate::test_support::database_url())
.await
.expect("test database");
let community = CommunityId::from_uuid(Uuid::new_v4());
sqlx::query("INSERT INTO communities (id, host) VALUES ($1, $2)")
.bind(community.as_uuid())
.bind(format!("admission-{}.example", community.as_uuid()))
.execute(&pool)
.await
.unwrap();
let owner = [0x31; 32];
crate::user::ensure_user(&pool, community, &owner)
.await
.unwrap();
let id = workflow::create_workflow(
&pool,
community,
None,
&owner,
"admission",
r#"{"enabled":true}"#,
&[0x41; 32],
)
.await
.unwrap();
let selected = workflow::get_workflow(&pool, community, id).await.unwrap();
(pool, selected)
}
#[tokio::test]
#[ignore = "requires Postgres"]
async fn admission_requires_selected_identity_revision_and_eligibility() {
let (pool, selected) = fixture().await;
let db = crate::Db::from_pool(pool.clone());
let trigger = serde_json::json!({"text":"original trigger"});
let event = [0x51; 32];
let id = db
.create_workflow_run(&selected, Some(&event), Some(&trigger))
.await
.unwrap()
.expect("current record admitted");
let run = db
.get_workflow_run(selected.community_id, id)
.await
.unwrap();
assert_eq!(run.workflow_id, selected.id);
assert_eq!(run.community_id, selected.community_id);
assert_eq!(run.trigger_event_id.as_deref(), Some(event.as_slice()));
assert_eq!(run.trigger_context, Some(trigger));
assert_eq!(run.status, workflow::RunStatus::Pending);
for field in ["community", "id", "owner", "channel", "hash", "incarnation"] {
let mut stale = selected.clone();
match field {
"community" => stale.community_id = CommunityId::from_uuid(Uuid::new_v4()),
"id" => stale.id = Uuid::new_v4(),
"owner" => stale.owner_pubkey = vec![0x32; 32],
"channel" => stale.channel_id = Some(Uuid::new_v4()),
"hash" => stale.definition_hash = vec![0x42; 32],
"incarnation" => stale.created_at += chrono::Duration::microseconds(1),
_ => unreachable!(),
}
assert!(
db.create_workflow_run(&stale, None, None)
.await
.unwrap()
.is_none(),
"{field}"
);
}
workflow::set_workflow_enabled(&pool, selected.community_id, selected.id, false)
.await
.unwrap();
assert!(db
.create_workflow_run(&selected, None, None)
.await
.unwrap()
.is_none());
workflow::set_workflow_enabled(&pool, selected.community_id, selected.id, true)
.await
.unwrap();
for status in [
workflow::WorkflowStatus::Disabled,
workflow::WorkflowStatus::Archived,
] {
workflow::update_workflow_status(&pool, selected.community_id, selected.id, status)
.await
.unwrap();
assert!(db
.create_workflow_run(&selected, None, None)
.await
.unwrap()
.is_none());
}
// Rejected attempts left no run rows, and did not alter an admitted run.
assert_eq!(
db.list_workflow_runs(selected.community_id, selected.id, 100)
.await
.unwrap()
.len(),
1
);
workflow::delete_workflow(&pool, selected.community_id, selected.id)
.await
.unwrap();
assert!(db
.create_workflow_run(&selected, None, None)
.await
.unwrap()
.is_none());
}
/// Observe a real server lock wait, not a sleep that merely hopes admission ran.
async fn wait_for_admission_lock(
pool: &PgPool,
writer_pid: i32,
admission: &tokio::task::JoinHandle<crate::Result<Option<Uuid>>>,
) {
tokio::time::timeout(Duration::from_secs(5), async {
loop {
assert!(!admission.is_finished(), "admission bypassed the in-flight workflow writer");
let blocked: bool = sqlx::query_scalar(
"SELECT EXISTS(SELECT 1 FROM pg_stat_activity WHERE datname = current_database() \
AND query LIKE '%WITH eligible AS%' AND $1 = ANY(pg_blocking_pids(pid)))",
).bind(writer_pid).fetch_one(pool).await.unwrap();
if blocked { return; }
tokio::time::sleep(Duration::from_millis(10)).await;
}
}).await.expect("admission must wait for the workflow row lock");
}
#[tokio::test]
#[ignore = "requires Postgres"]
async fn cluster_global_admission_waits_for_update_and_disable_then_rechecks() {
let (pool, selected) = fixture().await;
for mutation in [
"UPDATE workflows SET definition_hash = $3 WHERE community_id = $1 AND id = $2",
"UPDATE workflows SET enabled = FALSE WHERE community_id = $1 AND id = $2 AND definition_hash <> $3",
] {
let current = workflow::get_workflow(&pool, selected.community_id, selected.id).await.unwrap();
let mut tx = pool.begin().await.unwrap();
let pid: i32 = sqlx::query_scalar("SELECT pg_backend_pid()").fetch_one(&mut *tx).await.unwrap();
sqlx::query(mutation).bind(selected.community_id.as_uuid()).bind(selected.id).bind(vec![0x42u8; 32])
.execute(&mut *tx).await.unwrap();
let db = crate::Db::from_pool(pool.clone());
let admission = tokio::spawn(async move { db.create_workflow_run(&current, None, None).await });
wait_for_admission_lock(&pool, pid, &admission).await;
tx.commit().await.unwrap();
assert!(tokio::time::timeout(Duration::from_secs(5), admission).await.unwrap().unwrap().unwrap().is_none());
// Restore eligibility for the next independent in-flight mutation.
sqlx::query("UPDATE workflows SET definition_hash = $3, enabled = TRUE WHERE community_id = $1 AND id = $2")
.bind(selected.community_id.as_uuid()).bind(selected.id).bind(&selected.definition_hash)
.execute(&pool).await.unwrap();
}
assert!(
workflow::list_workflow_runs(&pool, selected.community_id, selected.id, 100)
.await
.unwrap()
.is_empty()
);
// A rolled-back disable must release the waiter and admit the unchanged record.
let mut tx = pool.begin().await.unwrap();
let pid: i32 = sqlx::query_scalar("SELECT pg_backend_pid()")
.fetch_one(&mut *tx)
.await
.unwrap();
sqlx::query("UPDATE workflows SET enabled = FALSE WHERE community_id = $1 AND id = $2")
.bind(selected.community_id.as_uuid())
.bind(selected.id)
.execute(&mut *tx)
.await
.unwrap();
let db = crate::Db::from_pool(pool.clone());
let admission =
tokio::spawn(async move { db.create_workflow_run(&selected, None, None).await });
wait_for_admission_lock(&pool, pid, &admission).await;
tx.rollback().await.unwrap();
assert!(tokio::time::timeout(Duration::from_secs(5), admission)
.await
.unwrap()
.unwrap()
.unwrap()
.is_some());
}
}

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

@@ -2180,9 +2180,10 @@ 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(&workflow, None, trigger_ctx_json.as_ref())
.await
.map_err(|e| super::internal_error(&format!("db error: {e}")))?;
.map_err(|e| super::internal_error(&format!("db error: {e}")))?
.ok_or_else(|| not_found("workflow not found"))?;
// Spawn workflow execution asynchronously.
let engine = Arc::clone(&state.workflow_engine);

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

@@ -953,14 +953,12 @@ async fn handle_workflow_trigger(
let event_id_bytes = event.id.as_bytes().to_vec();
let run_id = state
.db
.create_workflow_run(
community_id,
workflow_id,
Some(&event_id_bytes),
trigger_ctx_json.as_ref(),
)
.create_workflow_run(&workflow, Some(&event_id_bytes), trigger_ctx_json.as_ref())
.await
.map_err(|e| IngestError::Internal(format!("error: db create_workflow_run: {e}")))?;
.map_err(|e| IngestError::Internal(format!("error: db create_workflow_run: {e}")))?
.ok_or_else(|| {
IngestError::Rejected("forbidden: workflow changed or is no longer active".into())
})?;
// Finalize the idempotency record after the separate run creation succeeds.
tx.commit()
@@ -1675,3 +1673,6 @@ mod postgres_tests {
));
}
}
#[cfg(test)]
mod run_admission_postgres_tests;

259
crates/buzz-relay/src/handlers/command_executor/run_admission_postgres_tests.rs Normal file
View File

@@ -0,0 +1,259 @@
//! Production manual/webhook admission after a concurrent definition update.
use super::*;
use axum::{
extract::{Path, Query, State},
http::{HeaderMap, StatusCode},
};
use buzz_core::channel::{ChannelType, ChannelVisibility};
use nostr::{EventBuilder, Keys, Kind, Tag};
use std::{future::Future, pin::Pin, sync::Mutex, time::Duration};
#[derive(Default)]
struct Sink(Mutex<Vec<String>>);
impl buzz_workflow::ActionSink for Sink {
fn send_message(
&self,
_: CommunityId,
_: &str,
text: &str,
_: &str,
_: &str,
_: Option<&str>,
) -> Pin<Box<dyn Future<Output = Result<String, buzz_workflow::ActionSinkError>> + Send + '_>>
{
self.0.lock().unwrap().push(text.into());
Box::pin(async { Ok("ab".repeat(32)) })
}
}
async fn admission_race(webhook: bool) {
let state = crate::state::tests::test_state().await;
assert_eq!(
state.config.database_url,
std::env::var("BUZZ_TEST_DATABASE_URL").unwrap()
);
let pool = sqlx::PgPool::connect(&state.config.database_url)
.await
.unwrap();
let host = format!("admit-{}.example", Uuid::new_v4());
let community = state
.db
.ensure_configured_community(&host)
.await
.unwrap()
.id;
let tenant = TenantContext::resolved(community, host.clone());
let keys = Keys::generate();
state
.db
.ensure_user(community, keys.public_key().as_bytes())
.await
.unwrap();
let channel = state
.db
.create_channel(
community,
"admit",
ChannelType::Stream,
ChannelVisibility::Private,
None,
keys.public_key().as_bytes(),
None,
)
.await
.unwrap()
.id;
let id = Uuid::new_v4();
let mut definition = serde_json::json!({"name":"admit", "enabled":true,
"trigger":{"on":"webhook"}, "steps":[{"id":"emit","action":"send_message","text":"old-action"}]});
webhook_secret::inject_secret(&mut definition, "test-secret");
let json = definition.to_string();
state
.db
.upsert_workflow(
community,
id,
Some(channel),
keys.public_key().as_bytes(),
"admit",
&json,
&Sha256::digest(json.as_bytes()),
)
.await
.unwrap();
let sink = Arc::new(Sink::default());
state.workflow_engine.set_action_sink(sink.clone());
for commit_update in [true, false] {
// Block the real authority SELECT after the caller selected its definition.
// Only then update: a caller that refreshes the record but executes the
// original definition will now incorrectly admit old-action.
let mut tx = pool.begin().await.unwrap();
let pid: i32 = sqlx::query_scalar("SELECT pg_backend_pid()")
.fetch_one(&mut *tx)
.await
.unwrap();
sqlx::query("LOCK TABLE channel_members IN ACCESS EXCLUSIVE MODE")
.execute(&mut *tx)
.await
.unwrap();
let event = EventBuilder::new(
Kind::Custom(KIND_WORKFLOW_TRIGGER as u16),
Uuid::new_v4().to_string(),
)
.tags([Tag::parse(["d", &id.to_string()]).unwrap()])
.sign_with_keys(&keys)
.unwrap();
let event_id = event.id;
let call_state = state.clone();
let call_host = host.clone();
let call_tenant = tenant.clone();
let auth = IngestAuth::Nip42 {
pubkey: keys.public_key(),
scopes: vec![buzz_auth::Scope::MessagesWrite],
channel_ids: None,
conn_id: Uuid::new_v4(),
};
let call = tokio::spawn(async move {
if webhook {
let mut headers = HeaderMap::new();
headers.insert("host", call_host.parse().unwrap());
headers.insert("x-webhook-secret", "test-secret".parse().unwrap());
match crate::api::bridge::workflow_webhook(
State(call_state),
Path(id.to_string()),
Query(crate::api::bridge::WebhookQuery { secret: None }),
headers,
axum::body::Bytes::new(),
)
.await
{
Ok((code, _)) => {
assert_eq!(code, StatusCode::ACCEPTED);
true
}
Err((code, _)) => {
assert_eq!(code, StatusCode::NOT_FOUND);
false
}
}
} else {
match handle_workflow_trigger(&call_tenant, &call_state, &event, &auth).await {
Ok(result) => {
assert!(result.accepted);
true
}
Err(IngestError::Rejected(reason)) => {
assert_eq!(reason, "forbidden: workflow changed or is no longer active");
false
}
Err(e) => panic!("unexpected manual result: {e:?}"),
}
}
});
tokio::time::timeout(Duration::from_secs(5), async {
loop {
assert!(!call.is_finished(), "caller did not reach authority gate");
let waiting: bool = sqlx::query_scalar(
"SELECT EXISTS(SELECT 1 FROM pg_stat_activity \
WHERE datname=current_database() AND query LIKE '%channel_members%' \
AND $1=ANY(pg_blocking_pids(pid)))",
)
.bind(pid)
.fetch_one(&pool)
.await
.unwrap();
if waiting {
break;
}
tokio::time::sleep(Duration::from_millis(10)).await;
}
})
.await
.expect("authority SELECT must block");
definition["steps"][0]["text"] = serde_json::json!(if commit_update {
"new-action"
} else {
"rolled-back-action"
});
let json = definition.to_string();
buzz_db::workflow::upsert_workflow_in_transaction(
&mut tx,
community,
id,
Some(channel),
keys.public_key().as_bytes(),
"admit",
&json,
&Sha256::digest(json.as_bytes()),
)
.await
.unwrap();
if commit_update {
tx.commit().await.unwrap();
} else {
tx.rollback().await.unwrap();
}
let accepted = tokio::time::timeout(Duration::from_secs(5), call)
.await
.unwrap()
.unwrap();
assert_eq!(
accepted, !commit_update,
"stale selection must not admit; rollback must admit"
);
if commit_update {
assert!(state
.db
.list_workflow_runs(community, id, 100)
.await
.unwrap()
.is_empty());
assert!(sink.0.lock().unwrap().is_empty());
if !webhook {
let seen: bool = sqlx::query_scalar(
"SELECT EXISTS(SELECT 1 FROM events WHERE community_id=$1 AND id=$2)",
)
.bind(community.as_uuid())
.bind(event_id.as_bytes().as_slice())
.fetch_one(&pool)
.await
.unwrap();
assert!(!seen, "rejected manual command must roll back");
}
} else {
tokio::time::timeout(Duration::from_secs(5), async {
loop {
let runs = state
.db
.list_workflow_runs(community, id, 100)
.await
.unwrap();
assert_eq!(runs.len(), 1);
if runs[0].status == RunStatus::Completed {
break;
}
assert!(!matches!(
runs[0].status,
RunStatus::Failed | RunStatus::Cancelled
));
tokio::time::sleep(Duration::from_millis(10)).await;
}
})
.await
.expect("positive execution");
assert_eq!(*sink.0.lock().unwrap(), ["new-action"]);
}
}
}
#[tokio::test]
#[ignore = "requires PostgreSQL"]
async fn cluster_global_manual_admission_keeps_selected_revision_and_rejects_none() {
admission_race(false).await;
}
#[tokio::test]
#[ignore = "requires PostgreSQL"]
async fn cluster_global_webhook_admission_keeps_selected_revision_and_rejects_none() {
admission_race(true).await;
}

13
crates/buzz-relay/src/workflow_sink.rs
View File

@@ -862,12 +862,6 @@ mod postgres_tests {
..Default::default()
};
let trigger_ctx_json = serde_json::to_value(&trigger_ctx).expect("serialize trigger");
let run_id = state
.db
.create_workflow_run(community, workflow_id, None, Some(&trigger_ctx_json))
.await
.expect("create workflow run");
// Load the definition back from Postgres before execution. This pins the
// authority source to the durable owner-authored template rather than a
// second test-only string passed directly to RelayActionSink.
@@ -876,6 +870,13 @@ mod postgres_tests {
.get_workflow(community, workflow_id)
.await
.expect("load stored workflow");
let run_id = state
.db
.create_workflow_run(&stored_workflow, None, Some(&trigger_ctx_json))
.await
.expect("create workflow run")
.expect("current workflow admitted");
let stored_definition: buzz_workflow::WorkflowDef =
serde_json::from_value(stored_workflow.definition).expect("parse stored definition");
let result = buzz_workflow::executor::execute_run(

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

@@ -91,15 +91,10 @@ pub struct WorkflowEngine {
/// `(community_id, channel_id)`. Most channels have no workflows, so this
/// removes one SELECT from nearly every ingested event.
///
/// Consistency: the relay invalidates this cache on its own pod at the two
/// workflow mutation sites (command upsert, NIP-09 deletion). There is
/// deliberately no cross-pod invalidation — workflow triggering is not an
/// access-control fence, so the worst case on another pod is a just-deleted
/// workflow firing (or a just-created one missing events) for up to the TTL.
/// The same TTL also bounds the same-pod look-aside race (a stale fill
/// landing just after an invalidation). Workflow mutations are rare; the
/// 10s window matches the relay's other moka caches (see `AppState` in
/// `buzz-relay`).
/// Consistency: mutations invalidate the accepting pod's cache. Other pods
/// (and stale look-aside fills) may miss new triggers until the 10s TTL,
/// but cached records cannot admit superseded, disabled or recreated work:
/// run creation atomically fences the selected record against the live row.
pub(crate) workflow_cache:
moka::sync::Cache<(CommunityId, Uuid), Arc<Vec<buzz_db::workflow::WorkflowRecord>>>,
}
@@ -405,14 +400,17 @@ impl WorkflowEngine {
let run_id = match self
.db
.create_workflow_run(
community_id,
workflow.id,
workflow,
Some(&trigger_event_id_bytes),
Some(&trigger_ctx_json),
)
.await
{
Ok(id) => id,
Ok(Some(id)) => id,
Ok(None) => {
tracing::debug!(workflow_id = %workflow.id, "Skipping stale or inactive workflow");
continue;
}
Err(e) => {
tracing::error!(workflow_id = %workflow.id, "Failed to create run: {e}");
continue;
@@ -667,14 +665,17 @@ impl WorkflowEngine {
let run_id = match self
.db
.create_workflow_run(
community_id,
workflow.id,
workflow,
None, // no trigger event for cron
trigger_ctx_json.as_ref(),
)
.await
{
Ok(id) => id,
Ok(Some(id)) => id,
Ok(None) => {
tracing::debug!(workflow_id = %workflow.id, "Cron tick: skipping stale or inactive workflow");
continue;
}
Err(e) => {
tracing::error!(
workflow_id = %workflow.id,