From 2eb1fc279cba3bb9fd247887fbbcf1966b0a1c36 Mon Sep 17 00:00:00 2001 From: seiler Date: Mon, 5 Oct 2026 22:45:27 +0000 Subject: [PATCH 01/10] fix(workflow): block banned owners from webhook runs Signed-off-by: seiler Co-authored-by: Codex (cherry picked from commit 6ecd74515c01c407f817f7c25597337fcffa688a) Signed-off-by: Brad Seiler --- crates/buzz-relay/src/api/bridge.rs | 25 +- .../bridge/workflow_webhook_postgres_tests.rs | 358 ++++++++++++++++++ 2 files changed, 382 insertions(+), 1 deletion(-) create mode 100644 crates/buzz-relay/src/api/bridge/workflow_webhook_postgres_tests.rs diff --git a/crates/buzz-relay/src/api/bridge.rs b/crates/buzz-relay/src/api/bridge.rs index 2b8ff56152d..8d4e0df07bd 100644 --- a/crates/buzz-relay/src/api/bridge.rs +++ b/crates/buzz-relay/src/api/bridge.rs @@ -27,6 +27,10 @@ use super::{api_error, db_read_error, internal_error, not_found, parse_query_or_ mod thread_roots; mod thread_window; +#[cfg(test)] +#[path = "bridge/workflow_webhook_postgres_tests.rs"] +mod workflow_webhook_postgres_tests; + pub(crate) async fn enforce_http_admission( state: &AppState, tenant: &TenantContext, @@ -2394,7 +2398,11 @@ pub struct WebhookQuery { /// Webhook trigger endpoint. No user auth — the webhook secret authenticates the caller. /// /// Prefers `X-Webhook-Secret` header over `?secret=` query param (headers aren't logged -/// by most proxies). Returns 202 Accepted; execution is async. +/// by most proxies). Returns 202 Accepted; execution is async. A run is admitted +/// when its `workflow_runs` row is inserted; immediately before that insert, +/// admission checks the workflow owner's active ban in the host-bound community. +/// This point-in-time check does not cancel already-created runs, and a +/// concurrent ban committed after the lookup can race with row insertion. pub async fn workflow_webhook( State(state): State>, Path(id_str): Path, @@ -2510,6 +2518,21 @@ pub async fn workflow_webhook( .await .map_err(|_| not_found("workflow not found"))?; + // The workflow_runs insert is the admission boundary. This is the last + // authorization gate before it. The restriction read is authoritative and + // tenant-scoped; lookup failures deny admission just like an active ban. It + // is intentionally a point-in-time check rather than a transaction with ban + // writes, so a ban racing after this read may follow the run insertion. Runs + // already inserted are unaffected. + let restrictions = state + .db + .moderation_restriction_state(community_id, &workflow.owner_pubkey) + .await + .map_err(|_| not_found("workflow not found"))?; + if restrictions.banned { + return Err(not_found("workflow not found")); + } + let run_id = state .db .create_workflow_run(community_id, id, None, trigger_ctx_json.as_ref()) diff --git a/crates/buzz-relay/src/api/bridge/workflow_webhook_postgres_tests.rs b/crates/buzz-relay/src/api/bridge/workflow_webhook_postgres_tests.rs new file mode 100644 index 00000000000..2b60241ec3b --- /dev/null +++ b/crates/buzz-relay/src/api/bridge/workflow_webhook_postgres_tests.rs @@ -0,0 +1,358 @@ +use std::sync::Arc; + +use axum::{ + body::Bytes, + extract::{Path, Query, State}, + http::{header, HeaderMap, HeaderValue, StatusCode}, + Json, +}; +use serde_json::{json, Value}; +use sqlx::PgPool; +use uuid::Uuid; + +use super::{workflow_webhook, WebhookQuery}; +use crate::{ + config::Config, + state::{AppState, AuditShutdownHandle}, + test_support::database_url, +}; + +async fn app_state(db: buzz_db::Db, pool: &PgPool) -> (Arc, AuditShutdownHandle) { + let mut config = Config::for_test(); + config.database_url = database_url(); + config.redis_url = + std::env::var("REDIS_URL").expect("REDIS_URL must point to the isolated test Redis"); + config.relay_url = "wss://workflow-webhook-test.invalid".to_owned(); + config.require_auth_token = false; + config.require_relay_membership = false; + + let redis_pool = deadpool_redis::Config::from_url(&config.redis_url) + .create_pool(Some(deadpool_redis::Runtime::Tokio1)) + .expect("create Redis pool"); + let pubsub = Arc::new( + buzz_pubsub::PubSubManager::new(&config.redis_url, redis_pool.clone()) + .await + .expect("create pubsub manager"), + ); + let audit = buzz_audit::AuditService::new(pool.clone()); + let auth = buzz_auth::AuthService::new(config.auth.clone()); + let search = buzz_search::SearchService::new(pool.clone()); + let workflow_engine = Arc::new(buzz_workflow::WorkflowEngine::new( + db.clone(), + buzz_workflow::WorkflowConfig::default(), + )); + let media_storage = buzz_media::MediaStorage::new(&config.media).expect("create media storage"); + + let (state, audit_shutdown) = AppState::new( + config, + db, + redis_pool, + audit, + pubsub, + auth, + search, + workflow_engine, + nostr::Keys::generate(), + media_storage, + ); + (Arc::new(state), audit_shutdown) +} + +async fn seed_community(pool: &PgPool, community_id: Uuid, host: &str) { + sqlx::query("INSERT INTO communities (id, host) VALUES ($1, $2)") + .bind(community_id) + .bind(host) + .execute(pool) + .await + .expect("insert community"); +} + +struct WebhookWorkflowSeed<'a> { + community_id: Uuid, + workflow_id: Uuid, + channel_id: Uuid, + owner_pubkey: &'a [u8], + owner_role: &'a str, + name: &'a str, + secret: &'a str, + steps: Value, +} + +async fn seed_webhook_workflow(pool: &PgPool, seed: WebhookWorkflowSeed<'_>) { + sqlx::query("INSERT INTO users (community_id, pubkey) VALUES ($1, $2)") + .bind(seed.community_id) + .bind(seed.owner_pubkey) + .execute(pool) + .await + .expect("insert workflow owner"); + sqlx::query( + "INSERT INTO channels (community_id, id, name, created_by) VALUES ($1, $2, $3, $4)", + ) + .bind(seed.community_id) + .bind(seed.channel_id) + .bind(format!("workflow-{}", seed.name)) + .bind(seed.owner_pubkey) + .execute(pool) + .await + .expect("insert workflow channel"); + sqlx::query( + "INSERT INTO channel_members (community_id, channel_id, pubkey, role) \ + VALUES ($1, $2, $3, $4::member_role)", + ) + .bind(seed.community_id) + .bind(seed.channel_id) + .bind(seed.owner_pubkey) + .bind(seed.owner_role) + .execute(pool) + .await + .expect("insert workflow owner membership"); + + let definition = json!({ + "name": seed.name, + "trigger": {"on": "webhook"}, + "steps": seed.steps, + "_webhook_secret": seed.secret + }); + sqlx::query( + "INSERT INTO workflows \ + (community_id, id, name, owner_pubkey, channel_id, definition, definition_hash) \ + VALUES ($1, $2, $3, $4, $5, $6::jsonb, $7)", + ) + .bind(seed.community_id) + .bind(seed.workflow_id) + .bind(seed.name) + .bind(seed.owner_pubkey) + .bind(seed.channel_id) + .bind(definition.to_string()) + .bind(vec![0u8; 32]) + .execute(pool) + .await + .expect("insert workflow"); +} + +async fn post_webhook( + state: &Arc, + workflow_id: Uuid, + host: &str, + secret: &str, +) -> (StatusCode, Value) { + let mut headers = HeaderMap::new(); + headers.insert( + header::HOST, + HeaderValue::from_bytes(host.as_bytes()).expect("host header"), + ); + headers.insert( + "x-webhook-secret", + HeaderValue::from_bytes(secret.as_bytes()).expect("secret header"), + ); + + match workflow_webhook( + State(Arc::clone(state)), + Path(workflow_id.to_string()), + Query(WebhookQuery { secret: None }), + headers, + Bytes::new(), + ) + .await + { + Ok((status, Json(body))) | Err((status, Json(body))) => (status, body), + } +} + +async fn workflow_run_count(pool: &PgPool, community_id: Uuid, workflow_id: Uuid) -> i64 { + sqlx::query_scalar( + "SELECT COUNT(*) FROM workflow_runs WHERE community_id = $1 AND workflow_id = $2", + ) + .bind(community_id) + .bind(workflow_id) + .fetch_one(pool) + .await + .expect("count workflow runs") +} + +#[tokio::test] +#[ignore = "requires PostgreSQL and Redis"] +async fn webhook_admission_enforces_owner_ban_tenant_scope_and_lookup_failure() { + let pool = PgPool::connect(&database_url()) + .await + .expect("connect to isolated PostgreSQL test database"); + let db = buzz_db::Db::from_pool(pool.clone()); + + let community_a = Uuid::new_v4(); + let community_b = Uuid::new_v4(); + let host_a = format!("webhook-a-{}.test", Uuid::new_v4().simple()); + let host_b = format!("webhook-b-{}.test", Uuid::new_v4().simple()); + let owner_pubkey = [0x31; 32]; + let actor_pubkey = [0x52; 32]; + let member_pubkey = [0x73; 32]; + let workflow_id = Uuid::new_v4(); + let channel_id = Uuid::new_v4(); + let secret = format!("webhook-secret-{}", Uuid::new_v4().simple()); + + seed_community(&pool, community_a, &host_a).await; + seed_community(&pool, community_b, &host_b).await; + // Deliberately reuse workflow and channel IDs across tenants. The webhook + // request host must choose which community's owner restriction applies. + let delay_step = || { + json!([{ + "id": "hold", + "action": "delay", + "duration": "1s" + }]) + }; + seed_webhook_workflow( + &pool, + WebhookWorkflowSeed { + community_id: community_a, + workflow_id, + channel_id, + owner_pubkey: &owner_pubkey, + owner_role: "owner", + name: "owner-workflow-a", + secret: &secret, + steps: delay_step(), + }, + ) + .await; + seed_webhook_workflow( + &pool, + WebhookWorkflowSeed { + community_id: community_b, + workflow_id, + channel_id, + owner_pubkey: &owner_pubkey, + owner_role: "owner", + name: "owner-workflow-b", + secret: &secret, + steps: delay_step(), + }, + ) + .await; + + let member_workflow_id = Uuid::new_v4(); + let member_channel_id = Uuid::new_v4(); + seed_webhook_workflow( + &pool, + WebhookWorkflowSeed { + community_id: community_a, + workflow_id: member_workflow_id, + channel_id: member_channel_id, + owner_pubkey: &member_pubkey, + owner_role: "member", + name: "member-exfiltration-workflow", + secret: &secret, + steps: json!([{ + "id": "send", + "action": "call_webhook", + "url": "https://workflow-webhook-test.invalid" + }]), + }, + ) + .await; + + let (state, audit_shutdown) = app_state(db.clone(), &pool).await; + + let (status, _) = post_webhook(&state, workflow_id, &host_a, "wrong-secret").await; + assert_eq!( + status, + StatusCode::UNAUTHORIZED, + "the secret remains required" + ); + assert_eq!(workflow_run_count(&pool, community_a, workflow_id).await, 0); + + let (status, _) = post_webhook(&state, workflow_id, &host_a, &secret).await; + assert_eq!( + status, + StatusCode::ACCEPTED, + "an unbanned owner is admitted" + ); + assert_eq!(workflow_run_count(&pool, community_a, workflow_id).await, 1); + + db.ban_community_member( + buzz_core::CommunityId::from_uuid(community_a), + &owner_pubkey, + &actor_pubkey, + None, + None, + ) + .await + .expect("ban workflow owner"); + let (status, body) = post_webhook(&state, workflow_id, &host_a, &secret).await; + assert_eq!( + status, + StatusCode::NOT_FOUND, + "a banned owner is denied generically" + ); + assert_eq!(body["error"], "workflow not found"); + assert_eq!( + workflow_run_count(&pool, community_a, workflow_id).await, + 1, + "denial must not enqueue a workflow run" + ); + + let (status, _) = post_webhook(&state, workflow_id, &host_b, &secret).await; + assert_eq!( + status, + StatusCode::ACCEPTED, + "a ban in community A must not block the same owner in community B" + ); + assert_eq!(workflow_run_count(&pool, community_b, workflow_id).await, 1); + + assert!(db + .unban_community_member( + buzz_core::CommunityId::from_uuid(community_a), + &owner_pubkey, + &actor_pubkey, + ) + .await + .expect("unban workflow owner")); + let (status, _) = post_webhook(&state, workflow_id, &host_a, &secret).await; + assert_eq!( + status, + StatusCode::ACCEPTED, + "an explicitly unbanned owner is admitted" + ); + assert_eq!(workflow_run_count(&pool, community_a, workflow_id).await, 2); + + let (status, _) = post_webhook(&state, member_workflow_id, &host_a, &secret).await; + assert_eq!( + status, + StatusCode::NOT_FOUND, + "the existing elevated-role check still rejects member-owned exfiltration workflows" + ); + assert_eq!( + workflow_run_count(&pool, community_a, member_workflow_id).await, + 0, + "role denial must not enqueue a workflow run" + ); + + audit_shutdown + .drain(std::time::Duration::from_secs(1)) + .await; + drop(state); + + let (failing_db, admin_pool, schema) = + crate::test_support::restriction_lookup_failing_db().await; + let (failing_state, failing_audit_shutdown) = app_state(failing_db, &pool).await; + let (status, body) = post_webhook(&failing_state, workflow_id, &host_a, &secret).await; + assert_eq!( + status, + StatusCode::NOT_FOUND, + "restriction lookup failure denies admission" + ); + assert_eq!(body["error"], "workflow not found"); + assert_eq!( + workflow_run_count(&pool, community_a, workflow_id).await, + 2, + "a failed restriction lookup must not enqueue a workflow run" + ); + failing_audit_shutdown + .drain(std::time::Duration::from_secs(1)) + .await; + drop(failing_state); + + sqlx::raw_sql(sqlx::AssertSqlSafe(format!("DROP SCHEMA {schema} CASCADE"))) + .execute(&admin_pool) + .await + .expect("drop restriction lookup failure schema"); +} From adca0487b8fa2736f6c852fa4c342ace54c16605 Mon Sep 17 00:00:00 2001 From: Brad Seiler Date: Fri, 9 Oct 2026 19:26:04 -0400 Subject: [PATCH 02/10] fix(relay): retain mesh receives across housekeeping ticks Complete the preserved 66663e19 remediation with a deterministic production-consumer regression, cancellation-safety documentation, and deadline diagnostics. Original recovered refs and their attribution remain unchanged. Co-authored-by: Codex Signed-off-by: Brad Seiler (cherry picked from commit f1674a214188469c2420101df076ec5abc87ea5b) Signed-off-by: Brad Seiler --- crates/buzz-relay/src/api/mesh_demo.rs | 3 +- crates/buzz-relay/src/mesh_boot.rs | 109 +++++++++++++++++++--- crates/buzz-relay/src/tunnel/reliable.rs | 3 + crates/buzz-relay/tests/boot_lifecycle.rs | 52 +++++++---- 4 files changed, 133 insertions(+), 34 deletions(-) diff --git a/crates/buzz-relay/src/api/mesh_demo.rs b/crates/buzz-relay/src/api/mesh_demo.rs index 8649b97671a..77dc9aaed84 100644 --- a/crates/buzz-relay/src/api/mesh_demo.rs +++ b/crates/buzz-relay/src/api/mesh_demo.rs @@ -336,8 +336,9 @@ mod tests { }, ) .await; - assert_eq!(resp.status(), StatusCode::OK); + let status = resp.status(); let body = body_json(resp).await; + assert_eq!(status, StatusCode::OK, "forwarded response: {body}"); assert_eq!(body["outcome"], "forwarded"); assert_eq!(body["echoed_payload"], "mesh echo evidence"); owner_task.abort(); diff --git a/crates/buzz-relay/src/mesh_boot.rs b/crates/buzz-relay/src/mesh_boot.rs index 63a346258f5..d18d7ff8ca0 100644 --- a/crates/buzz-relay/src/mesh_boot.rs +++ b/crates/buzz-relay/src/mesh_boot.rs @@ -421,24 +421,39 @@ pub(crate) async fn run_demo_echo( tracing::info!(%session_id, %peer, "mesh demo echo: session open"); let mut drain_tick = tokio::time::interval(std::time::Duration::from_millis(100)); loop { - let frame = tokio::select! { - _ = drain_tick.tick() => { - if shutting_down.load(Ordering::Relaxed) { - if let Some(community_id) = stream.community_id() { - if let Err(e) = stream.send_goodbye(community_id, GoodbyeReason::Draining).await { - tracing::warn!(%session_id, "mesh demo echo: draining goodbye failed: {e}"); - } else { - tracing::info!(%session_id, "mesh demo echo: sent draining goodbye"); + let frame = { + // A receive can consume bytes before waiting on Redis validation. + // Keep it alive across housekeeping ticks; only shutdown may cancel it. + let recv = stream.recv_validated(&directory); + tokio::pin!(recv); + loop { + tokio::select! { + _ = drain_tick.tick() => { + if shutting_down.load(Ordering::Relaxed) { + break None; } - } else { - let _ = stream.finish(); - tracing::info!(%session_id, "mesh demo echo: drain before community latch — closing"); } - return; + frame = &mut recv => break Some(frame), } - continue; } - frame = stream.recv_validated(&directory) => frame, + }; + // The pending receive and its mutable stream borrow have ended before + // the drain response or next echo uses the stream. + let Some(frame) = frame else { + if let Some(community_id) = stream.community_id() { + if let Err(e) = stream + .send_goodbye(community_id, GoodbyeReason::Draining) + .await + { + tracing::warn!(%session_id, "mesh demo echo: draining goodbye failed: {e}"); + } else { + tracing::info!(%session_id, "mesh demo echo: sent draining goodbye"); + } + } else { + let _ = stream.finish(); + tracing::info!(%session_id, "mesh demo echo: drain before community latch — closing"); + } + return; }; match frame { Ok(Some(ReliableFrame::Data(payload))) => { @@ -696,6 +711,72 @@ mod tests { MeshStream::new(Box::new(StubSend), Box::new(StubRecv)) } + struct PausedRecv { + started: Option>, + resume: Option>, + calls: Arc, + } + + impl StreamRecvHalf for PausedRecv { + fn recv_frame(&mut self) -> BoxFuture<'_, Result, MeshError>> { + self.calls.fetch_add(1, Ordering::Relaxed); + let started = self.started.take(); + let resume = self.resume.take(); + Box::pin(async move { + if let Some(started) = started { + let _ = started.send(()); + } + if let Some(resume) = resume { + resume.await.expect("test resumes the in-flight receive"); + } + Ok(None) + }) + } + } + + /// Housekeeping must retain a receive that already consumed transport state. + /// Returning to recv_frame after a tick loses that state and closes the stream. + #[tokio::test(start_paused = true)] + async fn demo_echo_retains_pending_receive_across_housekeeping_ticks() { + let pool = deadpool_redis::Config::from_url("redis://127.0.0.1:1") + .create_pool(Some(deadpool_redis::Runtime::Tokio1)) + .expect("test pool"); + let (started_tx, started_rx) = tokio::sync::oneshot::channel(); + let (resume_tx, resume_rx) = tokio::sync::oneshot::channel(); + let calls = Arc::new(std::sync::atomic::AtomicUsize::new(0)); + let fenced = FencedHeader { + session_id: uuid::Uuid::new_v4(), + generation: 1, + owner_runtime_id: rid(1), + }; + let stream = MeshStream::new( + Box::new(StubSend), + Box::new(PausedRecv { + started: Some(started_tx), + resume: Some(resume_rx), + calls: Arc::clone(&calls), + }), + ); + let worker = tokio::spawn(run_demo_echo( + SessionDirectory::new(pool), + ReliableInbound { + fenced, + from: rid(2), + stream: crate::tunnel::reliable::ReliableMeshStream::new_inbound(fenced, stream), + }, + Arc::new(AtomicBool::new(false)), + )); + started_rx + .await + .expect("production consumer started receiving"); + tokio::time::advance(std::time::Duration::from_millis(300)).await; + tokio::task::yield_now().await; + assert_eq!(calls.load(Ordering::Relaxed), 1, "tick restarted a receive"); + assert!(!worker.is_finished(), "tick abandoned the pending receive"); + resume_tx.send(()).expect("receive remains alive"); + worker.await.expect("consumer completed after stream EOF"); + } + fn rid(byte: u8) -> RuntimeId { RuntimeId([byte; 32]) } diff --git a/crates/buzz-relay/src/tunnel/reliable.rs b/crates/buzz-relay/src/tunnel/reliable.rs index 5153e104df6..8238d410830 100644 --- a/crates/buzz-relay/src/tunnel/reliable.rs +++ b/crates/buzz-relay/src/tunnel/reliable.rs @@ -329,6 +329,9 @@ impl ReliableMeshStream { /// pinned fenced tuple and the Redis directory. This is the reliable-stream /// equivalent of Dawn's hot-path media floor, but authoritative: stale or /// mismatched frames fail the session rather than being dropped silently. + /// This receive is not cancellation-safe: it can consume frame bytes before + /// awaiting directory validation. Preserve an in-flight receive across + /// housekeeping ticks; cancellation is only safe when terminating the stream. pub async fn recv_validated( &mut self, directory: &SessionDirectory, diff --git a/crates/buzz-relay/tests/boot_lifecycle.rs b/crates/buzz-relay/tests/boot_lifecycle.rs index 027fc4d1c26..753200b47be 100644 --- a/crates/buzz-relay/tests/boot_lifecycle.rs +++ b/crates/buzz-relay/tests/boot_lifecycle.rs @@ -168,20 +168,38 @@ fn wait_for_relay_metrics(process: &mut RelayProcess, port: u16) -> String { /// [`METRICS_SCRAPE_DEADLINE`]. A relay that exits first is a failure, not a /// timeout, so the panic names the real cause. fn wait_for_scraped_metric(process: &mut RelayProcess, port: u16, needle: &str) -> String { + wait_for_scraped_metrics(process, port, &[needle]) +} + +/// Waits for one scrape to contain every requested sample, bounded by +/// [`METRICS_SCRAPE_DEADLINE`]. A scrape can overlap the sampler's sequential +/// metric writes, so related series may appear in different responses. +fn wait_for_scraped_metrics(process: &mut RelayProcess, port: u16, needles: &[&str]) -> String { + let requested = needles.join(", "); + let mut last_scrape = String::new(); + let mut last_error = None; let deadline = Instant::now() + METRICS_SCRAPE_DEADLINE; loop { assert!( process.try_wait().is_none(), - "relay exited before exporting {needle}" + "relay exited before exporting {requested}" ); - if let Ok(response) = scrape_metrics(port) { - if response.contains(needle) { - return response; + match scrape_metrics(port) { + Ok(response) => { + if needles + .iter() + .all(|needle| response.lines().any(|line| line.starts_with(needle))) + { + return response; + } + last_scrape = response; + last_error = None; } + Err(error) => last_error = Some(error.to_string()), } assert!( Instant::now() < deadline, - "relay did not export {needle} within {METRICS_SCRAPE_DEADLINE:?}" + "relay did not export {requested} within {METRICS_SCRAPE_DEADLINE:?}; last error: {last_error:?}; last scrape: {last_scrape}" ); thread::sleep(Duration::from_millis(20)); } @@ -814,10 +832,10 @@ mod postgres_tests { /// fires immediately, so the wait is bounded by /// [`METRICS_SCRAPE_DEADLINE`] and never a fixed sleep. /// - /// The completion-timestamp gauge is the needle because `sample` writes it - /// last, after the cache already serves the report: observing it proves the - /// whole cycle ran, with no window where the counters have landed but the - /// timestamp has not. + /// The sampler writes counters and histograms before the completion gauge, + /// but a concurrent `/metrics` scrape is not an atomic snapshot. Wait for + /// the actual timestamp, counter, and duration samples together so a scrape + /// that overlaps those sequential writes cannot report a false failure. #[test] #[ignore = "requires PostgreSQL"] fn startup_owns_the_dependency_sampler() { @@ -839,10 +857,14 @@ mod postgres_tests { ("REDIS_URL", &redis_url), ("BUZZ_GIT_CONFORMANCE_PROBE", "false"), ]); - let scrape = wait_for_scraped_metric( + let scrape = wait_for_scraped_metrics( &mut process, metrics_port, - "buzz_readiness_dependency_sample_completed_timestamp_seconds", + &[ + "buzz_readiness_dependency_sample_completed_timestamp_seconds ", + "buzz_readiness_dependency_checks_total{", + "buzz_readiness_check_duration_seconds_count{check=\"overall\"}", + ], ); let output = process.terminate(); let logs = format!( @@ -851,14 +873,6 @@ mod postgres_tests { String::from_utf8_lossy(&output.stderr) ); - assert!( - scrape.contains("buzz_readiness_dependency_checks_total{"), - "the same cycle must publish its per-dependency outcomes: {scrape}" - ); - assert!( - scrape.contains("buzz_readiness_check_duration_seconds"), - "a completed evaluation must publish its latency too: {scrape}" - ); assert!( !scrape.contains("buzz_readiness_checks_total{"), "no readiness probe was sent, so the sampler alone produced this: {scrape}" From 4b858d49d9752aa21b282f55fee63208c94a6287 Mon Sep 17 00:00:00 2001 From: Brad Seiler Date: Fri, 9 Oct 2026 19:29:04 -0400 Subject: [PATCH 03/10] test(relay): select mesh receive regression in both CI paths Co-authored-by: Codex Signed-off-by: Brad Seiler (cherry picked from commit 3d02a44a92bb095ff303c946c9aaf585f52c0b29) Signed-off-by: Brad Seiler --- Justfile | 2 ++ scripts/run-tests.sh | 3 +++ 2 files changed, 5 insertions(+) diff --git a/Justfile b/Justfile index 04c6e46e700..56bde2fb392 100644 --- a/Justfile +++ b/Justfile @@ -591,6 +591,8 @@ test-unit: + test(=state::tests::disconnect_community_wins_reason_losing_nip_fi_does_not_enqueue_frame) + test(=state::tests::manager_disconnect_sets_reason_enqueues_frame_then_cancels) + test(/^api::nip_fi::/)' + # Retain partially consumed mesh receives across housekeeping ticks. + cargo nextest run -p buzz-relay --lib -E 'test(=mesh_boot::tests::demo_echo_retains_pending_receive_across_housekeeping_ticks)' # boot_lifecycle spawns the real relay binary and asserts its startup # lifecycle, including that buzz_startup_phase_* reaches /metrics. Its # non-ignored tests need no Postgres or Redis; the Postgres cases are diff --git a/scripts/run-tests.sh b/scripts/run-tests.sh index 4f5ed92ba36..718ae82fce4 100755 --- a/scripts/run-tests.sh +++ b/scripts/run-tests.sh @@ -339,6 +339,9 @@ run_unit_tests() { run_test_step "buzz-relay binary tests" \ cargo test -p buzz-relay --bin buzz-relay -- --nocapture + run_test_step "buzz-relay mesh receive lifetime regression" \ + cargo test -p buzz-relay --lib mesh_boot::tests::demo_echo_retains_pending_receive_across_housekeeping_ticks -- --exact --nocapture + run_test_step "buzz-relay boot lifecycle tests" \ cargo test -p buzz-relay --test boot_lifecycle -- --nocapture } From 421d962726fa98d33778089629c3c3e79067f3b8 Mon Sep 17 00:00:00 2001 From: Brad Seiler Date: Fri, 9 Oct 2026 19:37:05 -0400 Subject: [PATCH 04/10] test(relay): initialize scraper errors from each observation Co-authored-by: Codex Signed-off-by: Brad Seiler (cherry picked from commit d43b957a5bfc4bd108e915be59db0a6af27d9798) Signed-off-by: Brad Seiler --- crates/buzz-relay/tests/boot_lifecycle.rs | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/crates/buzz-relay/tests/boot_lifecycle.rs b/crates/buzz-relay/tests/boot_lifecycle.rs index 753200b47be..ad6f20f17bd 100644 --- a/crates/buzz-relay/tests/boot_lifecycle.rs +++ b/crates/buzz-relay/tests/boot_lifecycle.rs @@ -177,7 +177,7 @@ fn wait_for_scraped_metric(process: &mut RelayProcess, port: u16, needle: &str) fn wait_for_scraped_metrics(process: &mut RelayProcess, port: u16, needles: &[&str]) -> String { let requested = needles.join(", "); let mut last_scrape = String::new(); - let mut last_error = None; + let mut last_error: Option; let deadline = Instant::now() + METRICS_SCRAPE_DEADLINE; loop { assert!( From d1c9dd341c3f8468ca976a99393ad1a7a0abefc6 Mon Sep 17 00:00:00 2001 From: Brad Seiler Date: Fri, 9 Oct 2026 20:25:54 -0400 Subject: [PATCH 05/10] test(db): derive push lease fixture time from persisted receipt Avoid comparing host and PostgreSQL clocks in the async enqueue guard regression. Assert strict lease ordering before releasing activation. Co-authored-by: Codex Signed-off-by: Brad Seiler (cherry picked from commit 893f81ffed0fa50048f786c7f698e9b8bf3a2ac6) Signed-off-by: Brad Seiler --- .../store/event_follow_up_postgres_tests.rs | 19 ++++++++++++++++++- 1 file changed, 18 insertions(+), 1 deletion(-) diff --git a/crates/buzz-db/src/store/event_follow_up_postgres_tests.rs b/crates/buzz-db/src/store/event_follow_up_postgres_tests.rs index b32b8bfbd44..107bbb38d5a 100644 --- a/crates/buzz-db/src/store/event_follow_up_postgres_tests.rs +++ b/crates/buzz-db/src/store/event_follow_up_postgres_tests.rs @@ -1018,13 +1018,30 @@ async fn assert_async_isolation(db: &Db, community: CommunityId, keys: &Keys) { endpoint_grant, max_class, subscriptions, expires_at, updated_at) \ SELECT $1, author, installation_id, source_event_id, source_created_at, generation, \ active, endpoint_enabled, endpoint_hash, endpoint_grant, max_class, \ - subscriptions, expires_at, clock_timestamp() FROM push_leases WHERE community_id=$2", + subscriptions, expires_at, \ + (SELECT received_at + interval '1 second' FROM events WHERE community_id=$1 AND id=$3) \ + FROM push_leases WHERE community_id=$2", ) .bind(later.as_uuid()) .bind(community.as_uuid()) + .bind(before_enrollment.id.as_bytes().as_slice()) .execute(&mut *activation) .await .unwrap(); + let leases_strictly_newer: bool = sqlx::query_scalar( + "SELECT count(*) > 0 AND bool_and(l.updated_at > e.received_at) \ + FROM push_leases l JOIN events e ON e.community_id=l.community_id \ + WHERE l.community_id=$1 AND e.id=$2", + ) + .bind(later.as_uuid()) + .bind(before_enrollment.id.as_bytes().as_slice()) + .fetch_one(&mut *activation) + .await + .unwrap(); + assert!( + leases_strictly_newer, + "fixture leases must follow recorded receipt" + ); activation.commit().await.unwrap(); assert_eq!(match_count(&isolated, later, &before_enrollment).await, 0); assert_eq!( From 9eb5e07d1b87f3904d1847c0b0e25e2191ef994f Mon Sep 17 00:00:00 2001 From: Luke Tornquist Date: Mon, 26 Oct 2026 14:05:51 -0400 Subject: [PATCH 06/10] fix(workflow): deny banned owners at every run-creation door The webhook handler checked the owner's ban itself, so cron and event-triggered workflows of a banned owner still created runs. Bans leave channel membership in place, so the shared SEC-006 owner authority gate passed them. Move the restriction read into WorkflowEngine::check_owner_authority, which the webhook, event, cron and signed-trigger doors all call before creating a run. A lookup failure still denies. The handler-local branch goes away; the webhook keeps its generic 404. Co-authored-by: Elrond <28d6302a099e5225b02c4155ac4236e4912603df2ab08dbfc2f4fef08ce598c8@buzz.block.builderlab.xyz> Co-Authored-By: Claude Opus 5.5 Signed-off-by: Luke Tornquist --- crates/buzz-relay/src/api/bridge.rs | 30 +++------ crates/buzz-workflow/src/lib.rs | 96 +++++++++++++++++++++++++++-- 2 files changed, 97 insertions(+), 29 deletions(-) diff --git a/crates/buzz-relay/src/api/bridge.rs b/crates/buzz-relay/src/api/bridge.rs index 36f0fa01191..8c416517103 100644 --- a/crates/buzz-relay/src/api/bridge.rs +++ b/crates/buzz-relay/src/api/bridge.rs @@ -2646,11 +2646,9 @@ pub struct WebhookQuery { /// Webhook trigger endpoint. No user auth — the webhook secret authenticates the caller. /// /// Prefers `X-Webhook-Secret` header over `?secret=` query param (headers aren't logged -/// by most proxies). Returns 202 Accepted; execution is async. A run is admitted -/// when its `workflow_runs` row is inserted; immediately before that insert, -/// admission checks the workflow owner's active ban in the host-bound community. -/// This point-in-time check does not cancel already-created runs, and a -/// concurrent ban committed after the lookup can race with row insertion. +/// by most proxies). Returns 202 Accepted; execution is async. Immediately +/// before the `workflow_runs` insert, the owner authority gate also rejects an +/// owner banned from the host-bound community. pub async fn workflow_webhook( State(state): State>, Path(id_str): Path, @@ -2744,9 +2742,10 @@ pub async fn workflow_webhook( // executes with the workflow **owner's** standing authority — so the // secret alone is insufficient. Immediately before run creation, reject // disabled/inactive workflows and recheck the owner's current channel - // membership (and role, for exfiltration-capable definitions). Fail - // closed with the same generic 404 as the lookups above so a - // revoked-owner workflow is indistinguishable from a nonexistent one. + // membership (and role, for exfiltration-capable definitions) and that + // the owner is not banned. Fail closed with the same generic 404 as the + // lookups above so a revoked-owner workflow is indistinguishable from a + // nonexistent one. if !workflow.enabled || workflow.status != buzz_db::workflow::WorkflowStatus::Active { return Err(not_found("workflow not found")); } @@ -2760,21 +2759,6 @@ pub async fn workflow_webhook( .await .map_err(|_| not_found("workflow not found"))?; - // The workflow_runs insert is the admission boundary. This is the last - // authorization gate before it. The restriction read is authoritative and - // tenant-scoped; lookup failures deny admission just like an active ban. It - // is intentionally a point-in-time check rather than a transaction with ban - // writes, so a ban racing after this read may follow the run insertion. Runs - // already inserted are unaffected. - let restrictions = state - .db - .moderation_restriction_state(community_id, &workflow.owner_pubkey) - .await - .map_err(|_| not_found("workflow not found"))?; - if restrictions.banned { - return Err(not_found("workflow not found")); - } - let run_id = state .db .create_workflow_run(community_id, id, None, trigger_ctx_json.as_ref()) diff --git a/crates/buzz-workflow/src/lib.rs b/crates/buzz-workflow/src/lib.rs index f33b57fdac9..bfe4b117b18 100644 --- a/crates/buzz-workflow/src/lib.rs +++ b/crates/buzz-workflow/src/lib.rs @@ -142,10 +142,15 @@ impl WorkflowEngine { /// in a role that can write (a guest cannot); /// - if the definition contains an exfiltration-capable action /// (`call_webhook`), the owner must currently hold the `owner` or - /// `admin` role. + /// `admin` role; + /// - the owner, or the human owner of an agent owner, must not be banned + /// from the community. Bans leave channel membership in place, so the + /// role check alone would keep a banned owner's workflows running. /// /// Any lookup error denies (fail-closed): a removed owner must never keep /// exfiltration authority because a membership read happened to fail. + /// This is a point-in-time check: it does not cancel runs that already + /// exist, and a ban that commits after it can race with run creation. pub async fn check_owner_authority( &self, community_id: CommunityId, @@ -162,13 +167,26 @@ impl WorkflowEngine { "owner authority lookup failed (fail-closed): {e}" )) })?; - if owner_authority_allows(role.as_deref(), def.requires_elevated_authority()) { - Ok(()) - } else { - Err(WorkflowError::Unauthorized( + if !owner_authority_allows(role.as_deref(), def.requires_elevated_authority()) { + return Err(WorkflowError::Unauthorized( "workflow owner lacks current channel authority".into(), - )) + )); } + let restrictions = self + .db + .moderation_restriction_state(community_id, owner_pubkey) + .await + .map_err(|e| { + WorkflowError::Unauthorized(format!( + "owner restriction lookup failed (fail-closed): {e}" + )) + })?; + if restrictions.banned { + return Err(WorkflowError::Unauthorized( + "workflow owner is banned from the community".into(), + )); + } + Ok(()) } /// Set the action sink. Called once after `AppState` construction. @@ -2037,6 +2055,72 @@ steps: ); } + /// A ban leaves the owner's channel membership in place, so the event + /// path must still stop creating runs while the owner is banned, and + /// resume once the ban is lifted. Cron and webhook runs pass through the + /// same `check_owner_authority` gate. + #[tokio::test] + #[ignore = "requires Postgres"] + async fn on_event_denies_run_while_owner_banned() { + let db = setup_db().await; + let creator = nostr::Keys::generate().public_key().to_bytes().to_vec(); + let member = nostr::Keys::generate().public_key().to_bytes().to_vec(); + let (community, channel_id) = setup_channel(&db, &creator, &member).await; + + let def_json = serde_json::json!({ + "name": "ban-event", + "trigger": {"on": "message_posted"}, + "steps": [{"id": "s1", "action": "send_message", "text": "hi"}], + "enabled": true, + }) + .to_string(); + let workflow_id = db + .create_workflow( + community, + Some(channel_id), + &member, + "ban-event", + &def_json, + &[0u8; 32], + ) + .await + .expect("create workflow"); + let engine = Arc::new(WorkflowEngine::new(db.clone(), WorkflowConfig::default())); + let run_count = || async { + db.list_workflow_runs(community, workflow_id, 10) + .await + .expect("list runs") + .len() + }; + + db.ban_community_member(community, &member, &creator, None, None) + .await + .expect("ban owner"); + engine + .on_event(community, &message_event(channel_id)) + .await + .expect("on_event while banned"); + assert_eq!( + run_count().await, + 0, + "a banned owner's workflow must not fire" + ); + + assert!(db + .unban_community_member(community, &member, &creator) + .await + .expect("unban owner")); + engine + .on_event(community, &message_event(channel_id)) + .await + .expect("on_event after unban"); + assert_eq!( + run_count().await, + 1, + "an unbanned owner's workflow fires again" + ); + } + /// Exfiltration-capable definitions (call_webhook) require the owner to /// currently hold an elevated role — a plain member's workflow must not /// fire even though the owner is still an active channel member. From 81c367f4815716f1c849e4dbe84775fa30537f57 Mon Sep 17 00:00:00 2001 From: Luke Tornquist Date: Mon, 26 Oct 2026 14:20:25 -0400 Subject: [PATCH 07/10] Expect generic 404 for wrong webhook secret in ban regression Main now answers a missing or invalid webhook secret with the same non-disclosing 404 as an unknown workflow. The merged Postgres regression still expected 401, so it stopped at its first assertion before exercising the ban, tenant-scope, unban and lookup-failure arms. Co-authored-by: Elrond <28d6302a099e5225b02c4155ac4236e4912603df2ab08dbfc2f4fef08ce598c8@buzz.block.builderlab.xyz> Co-Authored-By: Claude Opus 5.5 Signed-off-by: Luke Tornquist --- .../src/api/bridge/workflow_webhook_postgres_tests.rs | 7 ++++--- 1 file changed, 4 insertions(+), 3 deletions(-) diff --git a/crates/buzz-relay/src/api/bridge/workflow_webhook_postgres_tests.rs b/crates/buzz-relay/src/api/bridge/workflow_webhook_postgres_tests.rs index 2b60241ec3b..fb5cd7ffc00 100644 --- a/crates/buzz-relay/src/api/bridge/workflow_webhook_postgres_tests.rs +++ b/crates/buzz-relay/src/api/bridge/workflow_webhook_postgres_tests.rs @@ -252,12 +252,13 @@ async fn webhook_admission_enforces_owner_ban_tenant_scope_and_lookup_failure() let (state, audit_shutdown) = app_state(db.clone(), &pool).await; - let (status, _) = post_webhook(&state, workflow_id, &host_a, "wrong-secret").await; + let (status, body) = post_webhook(&state, workflow_id, &host_a, "wrong-secret").await; assert_eq!( status, - StatusCode::UNAUTHORIZED, - "the secret remains required" + StatusCode::NOT_FOUND, + "the secret remains required and is denied generically" ); + assert_eq!(body["error"], "workflow not found"); assert_eq!(workflow_run_count(&pool, community_a, workflow_id).await, 0); let (status, _) = post_webhook(&state, workflow_id, &host_a, &secret).await; From 1941b36b1aea3f326dce5fbb145c4daf6978ca43 Mon Sep 17 00:00:00 2001 From: Luke Tornquist Date: Thu, 29 Oct 2026 11:55:42 -0400 Subject: [PATCH 08/10] Rate-limit ban-denial warnings; pin cron, agent-owner and lookup-failure denials A banned owner's workflow is denied on every matching event and every cron tick, so the WARN it logged was steady noise for as long as the ban lasted. The gate now returns a typed `OwnerBanned` denial, and the event and cron paths log it at WARN at most once per workflow per ten minutes (DEBUG in between). Every other denial, including a failed lookup, still warns. The cron loop body moves into `cron_tick(now)` so a test can drive one tick. New Postgres tests: - a cron tick skips a banned owner and an agent whose human owner is banned without consuming either scheduled-fire claim, while an unbanned workflow on the same tick still fires; - the event path fails closed when the restriction lookup errors. buzz-workflow's infra-free unit tests ran in no CI lane; `just test-unit` (and its cargo fallback) now runs them. Co-authored-by: Elrond <28d6302a099e5225b02c4155ac4236e4912603df2ab08dbfc2f4fef08ce598c8@buzz.block.builderlab.xyz> Co-Authored-By: Claude Opus 5.5 Signed-off-by: Luke Tornquist --- Justfile | 3 + crates/buzz-workflow/Cargo.toml | 3 + crates/buzz-workflow/src/error.rs | 7 + crates/buzz-workflow/src/lib.rs | 754 ++++++++++++++++++++---------- scripts/run-tests.sh | 4 + 5 files changed, 535 insertions(+), 236 deletions(-) diff --git a/Justfile b/Justfile index b803b8e57d5..e7dfbf7d007 100644 --- a/Justfile +++ b/Justfile @@ -429,6 +429,9 @@ test-unit: # invocation hits. Enumerated explicitly because nothing in CI runs # `cargo test --workspace` — membership buys clippy/check, not tests. cargo nextest run -p buzz-sdk --lib + # buzz-workflow parser, cron, condition and denial-logging unit tests + # (infra-free). Its ignored Postgres cases run in the PostgreSQL lane. + cargo nextest run -p buzz-workflow --lib # buzz-acp owns the relay-to-agent trust boundary. Run its tests here so # forged relay events cannot regain a path into agent routing unnoticed. cargo nextest run -p buzz-acp diff --git a/crates/buzz-workflow/Cargo.toml b/crates/buzz-workflow/Cargo.toml index 7d361b1477a..70a3a617905 100644 --- a/crates/buzz-workflow/Cargo.toml +++ b/crates/buzz-workflow/Cargo.toml @@ -27,5 +27,8 @@ tracing = { workspace = true } thiserror = { workspace = true } reqwest = { workspace = true, optional = true } +[dev-dependencies] +sqlx = { workspace = true } + [features] reqwest = ["dep:reqwest"] diff --git a/crates/buzz-workflow/src/error.rs b/crates/buzz-workflow/src/error.rs index 109d4a2cb3d..b563f51ebb5 100644 --- a/crates/buzz-workflow/src/error.rs +++ b/crates/buzz-workflow/src/error.rs @@ -60,6 +60,12 @@ pub enum WorkflowError { #[error("unauthorized: {0}")] Unauthorized(String), + /// The workflow's owner, or the human owner of an agent owner, is banned + /// from the community. An expected, repeating denial for as long as the + /// ban lasts, so callers may log it at a lower rate than `Unauthorized`. + #[error("unauthorized: workflow owner is banned from the community")] + OwnerBanned, + /// The action is defined but not yet implemented. #[error("action not implemented: {0}")] NotImplemented(String), @@ -78,6 +84,7 @@ impl WorkflowError { Self::CapacityExceeded => "capacity_exceeded", Self::Database(_) => "database_error", Self::Unauthorized(_) => "owner_unauthorized", + Self::OwnerBanned => "owner_banned", Self::NotImplemented(_) => "action_not_implemented", } } diff --git a/crates/buzz-workflow/src/lib.rs b/crates/buzz-workflow/src/lib.rs index bfe4b117b18..a515af4b8b0 100644 --- a/crates/buzz-workflow/src/lib.rs +++ b/crates/buzz-workflow/src/lib.rs @@ -102,8 +102,17 @@ pub struct WorkflowEngine { /// `buzz-relay`). pub(crate) workflow_cache: moka::sync::Cache<(CommunityId, Uuid), Arc>>, + /// Workflows whose ban denial was logged at WARN within the last + /// [`BAN_DENIAL_WARN_INTERVAL`], keyed `(community_id, workflow_id)`. A + /// banned owner's workflow is denied on every matching event and every + /// cron tick; this keeps that expected denial from flooding the log. + pub(crate) ban_denial_warned: moka::sync::Cache<(CommunityId, Uuid), ()>, } +/// How often a banned owner's workflow denial is logged at WARN. Denials in +/// between log at DEBUG. Other authority failures always log at WARN. +const BAN_DENIAL_WARN_INTERVAL: std::time::Duration = std::time::Duration::from_secs(600); + impl WorkflowEngine { /// Create a new `WorkflowEngine`. pub fn new(db: Db, config: WorkflowConfig) -> Self { @@ -119,6 +128,10 @@ impl WorkflowEngine { .max_capacity(10_000) .time_to_live(std::time::Duration::from_secs(10)) .build(), + ban_denial_warned: moka::sync::Cache::builder() + .max_capacity(10_000) + .time_to_live(BAN_DENIAL_WARN_INTERVAL) + .build(), } } @@ -182,13 +195,37 @@ impl WorkflowEngine { )) })?; if restrictions.banned { - return Err(WorkflowError::Unauthorized( - "workflow owner is banned from the community".into(), - )); + return Err(WorkflowError::OwnerBanned); } Ok(()) } + /// Log a [`check_owner_authority`](Self::check_owner_authority) denial + /// that skipped a run. A ban denial logs at WARN at most once per + /// workflow per [`BAN_DENIAL_WARN_INTERVAL`], and at DEBUG otherwise; + /// every other denial, including a failed lookup, logs at WARN. + fn log_owner_denial( + &self, + community_id: CommunityId, + workflow_id: Uuid, + trigger: &'static str, + err: &WorkflowError, + ) { + if should_warn_owner_denial(&self.ban_denial_warned, community_id, workflow_id, err) { + tracing::warn!( + workflow_id = %workflow_id, + trigger, + "Skipping workflow — owner authority check failed: {err}" + ); + } else { + tracing::debug!( + workflow_id = %workflow_id, + trigger, + "Skipping workflow — owner authority check failed: {err}" + ); + } + } + /// Set the action sink. Called once after `AppState` construction. /// /// # Panics @@ -413,10 +450,7 @@ impl WorkflowEngine { .check_owner_authority(community_id, channel_id, &workflow.owner_pubkey, &def) .await { - tracing::warn!( - workflow_id = %workflow.id, - "Skipping workflow — owner authority check failed: {e}" - ); + self.log_owner_denial(community_id, workflow.id, "event", &e); continue; } @@ -511,261 +545,275 @@ impl WorkflowEngine { loop { tokio::time::sleep(std::time::Duration::from_secs(60)).await; - let now = Utc::now(); - - let workflows = match self.db.list_all_enabled_workflows().await { - Ok(wf) => wf, - Err(e) => { - tracing::error!("Cron tick: failed to load workflows: {e}"); - continue; - } - }; - - for workflow in &workflows { - // The same workflow UUID may exist in another community; carry - // the row's owning community through fire-tracking, run creation, - // and execution so a fire/run never crosses tenants. - let community_id = workflow.community_id; - let def: schema::WorkflowDef = - match serde_json::from_value(workflow.definition.clone()) { - Ok(d) => d, - Err(e) => { - tracing::warn!( - workflow_id = %workflow.id, - "Cron tick: failed to parse workflow definition: {e}" - ); - continue; - } - }; + self.cron_tick(Utc::now()).await; + } + } - if !def.enabled { - continue; - } + /// One pass of the cron loop at `now`: fire every enabled schedule + /// workflow that is due, then prune the interval clock. + pub(crate) async fn cron_tick(self: &Arc, now: DateTime) { + let workflows = match self.db.list_all_enabled_workflows().await { + Ok(wf) => wf, + Err(e) => { + tracing::error!("Cron tick: failed to load workflows: {e}"); + return; + } + }; - // Fix 2: skip workflows with no channel_id — an empty channel_id - // causes silent downstream failures when the run tries to act on a channel. - let Some(channel_id) = workflow.channel_id else { - tracing::warn!( - workflow_id = %workflow.id, - "Cron tick: skipping schedule workflow with no channel_id" - ); - continue; - }; - - // Resolve the *deterministic* schedule instant this tick is - // firing for. `scheduled_for` is computed identically on every - // pod (cron's own scheduled time, or the interval bucket - // boundary) so all pods collide on a single durable claim — - // never `now`, which is per-pod and would let every pod fire. - let (scheduled_for, trigger_type) = match &def.trigger { - schema::TriggerDef::Schedule { - cron: Some(expr), - interval: None, - } => match cron_fire_instant(expr, now, 60, workflow.id) { - Some(instant) => (instant, "cron"), - None => continue, - }, - schema::TriggerDef::Schedule { - cron: None, - interval: Some(dur), - } => { - // Cheap pre-filter: skip the claim attempt when the - // in-memory clock says we're clearly mid-interval. The - // durable claim below is the real at-most-once boundary; - // this only avoids a DB write every tick. Seed the - // anchor from the DB on the first tick after restart so - // a process bounce can't double-fire within an interval. - let last = match self.last_fired.get(&(community_id, workflow.id)) { - Some(t) => Some(*t), - None => match self - .db - .latest_scheduled_workflow_fire(community_id, workflow.id) - .await - { - Ok(anchor) => anchor, - Err(e) => { - // Fail closed: a missing anchor reads as - // last_fired = now in interval_should_fire, - // so this tick is suppressed and the next - // tick retries. Surface the read failure so - // a persistently-unreadable anchor is visible - // rather than silently stalling the schedule. - tracing::warn!( - community_id = %community_id, - workflow_id = %workflow.id, - "Cron tick: failed to read interval restart anchor, \ - suppressing this tick: {e}" - ); - None - } - }, - }; - if !self.interval_prefilter_should_fire( - community_id, - workflow.id, - dur, - last, - now, - ) { - continue; - } - match interval_fire_instant(dur, now, workflow.id) { - Some(instant) => (instant, "interval"), - None => continue, - } - } - _ => continue, // Non-schedule triggers handled by on_event() - }; - - // SEC-006: recheck the owner's current channel authority - // BEFORE the durable claim. Placing the gate after the claim - // would let a revoked owner's workflow consume the - // at-most-once fire slot (claims are never re-fired), turning - // revocation into a denial-of-fire for a later re-enable. - if let Err(e) = self - .check_owner_authority(community_id, channel_id, &workflow.owner_pubkey, &def) - .await - { + for workflow in &workflows { + // The same workflow UUID may exist in another community; carry + // the row's owning community through fire-tracking, run creation, + // and execution so a fire/run never crosses tenants. + let community_id = workflow.community_id; + let def: schema::WorkflowDef = match serde_json::from_value(workflow.definition.clone()) + { + Ok(d) => d, + Err(e) => { tracing::warn!( workflow_id = %workflow.id, - "Cron tick: skipping workflow — owner authority check failed: {e}" + "Cron tick: failed to parse workflow definition: {e}" ); continue; } + }; - // Durable at-most-once claim — the cross-pod fire boundary. - // The loser receives `None` and skips BEFORE any run creation or - // side effect. `community_id` is the workflow row's own - // community (server provenance from the scan), never client - // input; the claim binds `(community_id, workflow_id, - // scheduled_for)` so a duplicate workflow UUID in another - // community claims independently. - match self - .db - .claim_scheduled_workflow_fire(community_id, workflow.id, scheduled_for) - .await - { - Ok(Some(_)) => {} - Ok(None) => { - // Another pod (or an earlier tick this pod) already - // claimed this instant. Still advance the in-memory - // interval clock so we don't re-attempt the claim every - // tick for the rest of the interval. - if trigger_type == "interval" { - self.last_fired.insert((community_id, workflow.id), now); - } + if !def.enabled { + continue; + } + + // Fix 2: skip workflows with no channel_id — an empty channel_id + // causes silent downstream failures when the run tries to act on a channel. + let Some(channel_id) = workflow.channel_id else { + tracing::warn!( + workflow_id = %workflow.id, + "Cron tick: skipping schedule workflow with no channel_id" + ); + continue; + }; + + // Resolve the *deterministic* schedule instant this tick is + // firing for. `scheduled_for` is computed identically on every + // pod (cron's own scheduled time, or the interval bucket + // boundary) so all pods collide on a single durable claim — + // never `now`, which is per-pod and would let every pod fire. + let (scheduled_for, trigger_type) = match &def.trigger { + schema::TriggerDef::Schedule { + cron: Some(expr), + interval: None, + } => match cron_fire_instant(expr, now, 60, workflow.id) { + Some(instant) => (instant, "cron"), + None => continue, + }, + schema::TriggerDef::Schedule { + cron: None, + interval: Some(dur), + } => { + // Cheap pre-filter: skip the claim attempt when the + // in-memory clock says we're clearly mid-interval. The + // durable claim below is the real at-most-once boundary; + // this only avoids a DB write every tick. Seed the + // anchor from the DB on the first tick after restart so + // a process bounce can't double-fire within an interval. + let last = match self.last_fired.get(&(community_id, workflow.id)) { + Some(t) => Some(*t), + None => match self + .db + .latest_scheduled_workflow_fire(community_id, workflow.id) + .await + { + Ok(anchor) => anchor, + Err(e) => { + // Fail closed: a missing anchor reads as + // last_fired = now in interval_should_fire, + // so this tick is suppressed and the next + // tick retries. Surface the read failure so + // a persistently-unreadable anchor is visible + // rather than silently stalling the schedule. + tracing::warn!( + community_id = %community_id, + workflow_id = %workflow.id, + "Cron tick: failed to read interval restart anchor, \ + suppressing this tick: {e}" + ); + None + } + }, + }; + if !self.interval_prefilter_should_fire( + community_id, + workflow.id, + dur, + last, + now, + ) { continue; } - Err(e) => { - tracing::error!( - workflow_id = %workflow.id, - "Cron tick: scheduled-fire claim failed: {e}" - ); - continue; + match interval_fire_instant(dur, now, workflow.id) { + Some(instant) => (instant, "interval"), + None => continue, } } + _ => continue, // Non-schedule triggers handled by on_event() + }; - // Fix 5: handle serialization errors explicitly rather than silently - // dropping the trigger context with .ok(). - let trigger_ctx = executor::TriggerContext { - channel_id: channel_id.to_string(), - timestamp: now.timestamp().to_string(), - ..Default::default() - }; - let trigger_ctx_json = match serde_json::to_value(&trigger_ctx) { - Ok(v) => Some(v), - Err(e) => { - tracing::error!( - workflow_id = %workflow.id, - "Cron tick: failed to serialize trigger context: {e}" - ); - continue; - } - }; + // SEC-006: recheck the owner's current channel authority + // BEFORE the durable claim. Placing the gate after the claim + // would let a revoked owner's workflow consume the + // at-most-once fire slot (claims are never re-fired), turning + // revocation into a denial-of-fire for a later re-enable. + if let Err(e) = self + .check_owner_authority(community_id, channel_id, &workflow.owner_pubkey, &def) + .await + { + self.log_owner_denial(community_id, workflow.id, trigger_type, &e); + continue; + } - let run_id = match self - .db - .create_workflow_run( - community_id, - workflow.id, - None, // no trigger event for cron - trigger_ctx_json.as_ref(), - ) - .await - { - Ok(id) => id, - Err(e) => { - tracing::error!( - workflow_id = %workflow.id, - "Cron tick: failed to create workflow run: {e}" - ); - // The claim is held but the run failed to create. The - // claim row intentionally stays (its `workflow_run_id` - // NULL) so this instant is not re-fired: at-most-once is - // preserved over exactly-once on transient run-insert - // failures. - continue; + // Durable at-most-once claim — the cross-pod fire boundary. + // The loser receives `None` and skips BEFORE any run creation or + // side effect. `community_id` is the workflow row's own + // community (server provenance from the scan), never client + // input; the claim binds `(community_id, workflow_id, + // scheduled_for)` so a duplicate workflow UUID in another + // community claims independently. + match self + .db + .claim_scheduled_workflow_fire(community_id, workflow.id, scheduled_for) + .await + { + Ok(Some(_)) => {} + Ok(None) => { + // Another pod (or an earlier tick this pod) already + // claimed this instant. Still advance the in-memory + // interval clock so we don't re-attempt the claim every + // tick for the rest of the interval. + if trigger_type == "interval" { + self.last_fired.insert((community_id, workflow.id), now); } - }; + continue; + } + Err(e) => { + tracing::error!( + workflow_id = %workflow.id, + "Cron tick: scheduled-fire claim failed: {e}" + ); + continue; + } + } - // Link the won claim to its run for ops/audit forensics. The - // claim row already guarantees dedupe; this is best-effort. - if let Err(e) = self - .db - .attach_scheduled_workflow_run(community_id, workflow.id, scheduled_for, run_id) - .await - { - tracing::warn!( + // Fix 5: handle serialization errors explicitly rather than silently + // dropping the trigger context with .ok(). + let trigger_ctx = executor::TriggerContext { + channel_id: channel_id.to_string(), + timestamp: now.timestamp().to_string(), + ..Default::default() + }; + let trigger_ctx_json = match serde_json::to_value(&trigger_ctx) { + Ok(v) => Some(v), + Err(e) => { + tracing::error!( workflow_id = %workflow.id, - run_id = %run_id, - "Cron tick: failed to attach run to scheduled-fire claim: {e}" + "Cron tick: failed to serialize trigger context: {e}" ); + continue; } + }; - // Update last_fired AFTER a successful claim+insert so that a - // failure doesn't suppress the next tick for the full interval. - // Only needed for interval triggers — cron uses window-based - // matching which already prevents double-fire within the same - // minute, and the durable claim backstops both. - if trigger_type == "interval" { - self.last_fired.insert((community_id, workflow.id), now); + let run_id = match self + .db + .create_workflow_run( + community_id, + workflow.id, + None, // no trigger event for cron + trigger_ctx_json.as_ref(), + ) + .await + { + Ok(id) => id, + Err(e) => { + tracing::error!( + workflow_id = %workflow.id, + "Cron tick: failed to create workflow run: {e}" + ); + // The claim is held but the run failed to create. The + // claim row intentionally stays (its `workflow_run_id` + // NULL) so this instant is not re-fired: at-most-once is + // preserved over exactly-once on transient run-insert + // failures. + continue; } + }; - // Fix 6: log the specific trigger type (cron vs interval). - tracing::info!( + // Link the won claim to its run for ops/audit forensics. The + // claim row already guarantees dedupe; this is best-effort. + if let Err(e) = self + .db + .attach_scheduled_workflow_run(community_id, workflow.id, scheduled_for, run_id) + .await + { + tracing::warn!( workflow_id = %workflow.id, run_id = %run_id, - trigger = trigger_type, - "Cron trigger fired" + "Cron tick: failed to attach run to scheduled-fire claim: {e}" ); + } - let engine = Arc::clone(self); - let def_clone = def.clone(); - let ctx_clone = trigger_ctx.clone(); - tokio::spawn(async move { - let result = executor::execute_run( - &engine, - community_id, - run_id, - &def_clone, - &ctx_clone, - ) - .await; - engine - .finalize_run(community_id, run_id, result, None) - .await; - }); + // Update last_fired AFTER a successful claim+insert so that a + // failure doesn't suppress the next tick for the full interval. + // Only needed for interval triggers — cron uses window-based + // matching which already prevents double-fire within the same + // minute, and the durable claim backstops both. + if trigger_type == "interval" { + self.last_fired.insert((community_id, workflow.id), now); } - // Fix 1: prune stale last_fired entries for workflows that are no longer - // active/enabled. Without this the DashMap grows monotonically as - // workflows are deleted or disabled. Keyed by `(community_id, id)` so - // entries are matched to the same scope they were inserted under. - let active_ids: std::collections::HashSet<(CommunityId, Uuid)> = - workflows.iter().map(|w| (w.community_id, w.id)).collect(); - self.last_fired.retain(|key, _| active_ids.contains(key)); + // Fix 6: log the specific trigger type (cron vs interval). + tracing::info!( + workflow_id = %workflow.id, + run_id = %run_id, + trigger = trigger_type, + "Cron trigger fired" + ); + + let engine = Arc::clone(self); + let def_clone = def.clone(); + let ctx_clone = trigger_ctx.clone(); + tokio::spawn(async move { + let result = + executor::execute_run(&engine, community_id, run_id, &def_clone, &ctx_clone) + .await; + engine + .finalize_run(community_id, run_id, result, None) + .await; + }); } + + // Fix 1: prune stale last_fired entries for workflows that are no longer + // active/enabled. Without this the DashMap grows monotonically as + // workflows are deleted or disabled. Keyed by `(community_id, id)` so + // entries are matched to the same scope they were inserted under. + let active_ids: std::collections::HashSet<(CommunityId, Uuid)> = + workflows.iter().map(|w| (w.community_id, w.id)).collect(); + self.last_fired.retain(|key, _| active_ids.contains(key)); + } +} + +/// Whether an owner-authority denial for this workflow logs at WARN. A ban +/// denial warns only when no warning for the same workflow is still inside +/// `warned`'s TTL; every other denial always warns. +fn should_warn_owner_denial( + warned: &moka::sync::Cache<(CommunityId, Uuid), ()>, + community_id: CommunityId, + workflow_id: Uuid, + err: &WorkflowError, +) -> bool { + if !matches!(err, WorkflowError::OwnerBanned) { + return true; } + warned + .entry((community_id, workflow_id)) + .or_insert(()) + .is_fresh() } /// Find the cron schedule instant that fired within the `window_secs`-wide @@ -1922,16 +1970,19 @@ steps: // -- SEC-006: event-path regression (requires Postgres) ---------------- - async fn setup_db() -> buzz_db::Db { - let database_url = std::env::var("BUZZ_TEST_DATABASE_URL") + fn test_database_url() -> String { + std::env::var("BUZZ_TEST_DATABASE_URL") .or_else(|_| std::env::var("DATABASE_URL")) // Local-only test default; this is not a production credential. .unwrap_or_else(|_| { let local_test_database = "postgres://buzz:buzz_dev@localhost:5432/buzz"; // sadscan:disable np.postgres.1 local_test_database.to_owned() - }); + }) + } + + async fn setup_db() -> buzz_db::Db { buzz_db::Db::new(&buzz_db::DbConfig { - database_url, + database_url: test_database_url(), ..Default::default() }) .await @@ -2188,4 +2239,235 @@ steps: "channel owner's call_webhook workflow fires" ); } + + #[test] + fn ban_denial_warns_once_per_interval_while_other_denials_always_warn() { + let warned = moka::sync::Cache::builder() + .max_capacity(16) + .time_to_live(BAN_DENIAL_WARN_INTERVAL) + .build(); + let community = CommunityId::from_uuid(Uuid::new_v4()); + let (wf_a, wf_b) = (Uuid::new_v4(), Uuid::new_v4()); + let banned = WorkflowError::OwnerBanned; + let lookup_failed = WorkflowError::Unauthorized( + "owner restriction lookup failed (fail-closed): boom".into(), + ); + + assert!(should_warn_owner_denial(&warned, community, wf_a, &banned)); + assert!( + !should_warn_owner_denial(&warned, community, wf_a, &banned), + "a repeated ban denial inside the interval logs below WARN" + ); + assert!( + should_warn_owner_denial(&warned, community, wf_b, &banned), + "each workflow gets its own first warning" + ); + assert!( + should_warn_owner_denial(&warned, community, wf_a, &lookup_failed), + "a failed lookup always warns" + ); + assert!( + should_warn_owner_denial(&warned, community, wf_a, &lookup_failed), + "a failed lookup is never rate-limited" + ); + } + + fn schedule_def(name: &str) -> String { + serde_json::json!({ + "name": name, + "trigger": {"on": "schedule", "cron": "* * * * *"}, + "steps": [{"id": "s1", "action": "send_message", "text": "tick"}], + "enabled": true, + }) + .to_string() + } + + /// A cron tick must skip a banned owner's workflow and an agent-owned + /// workflow whose human owner is banned, without consuming either + /// workflow's scheduled-fire claim: once unbanned, a tick for the same + /// instant fires them. An unbanned workflow on the same tick still fires. + #[tokio::test] + #[ignore = "requires Postgres"] + async fn cron_tick_skips_banned_owners_without_consuming_the_claim() { + let db = setup_db().await; + let creator = nostr::Keys::generate().public_key().to_bytes().to_vec(); + let member = nostr::Keys::generate().public_key().to_bytes().to_vec(); + let (community, channel_id) = setup_channel(&db, &creator, &member).await; + + let human = nostr::Keys::generate().public_key().to_bytes().to_vec(); + let agent = nostr::Keys::generate().public_key().to_bytes().to_vec(); + db.ensure_user(community, &human).await.expect("human user"); + db.ensure_user(community, &agent).await.expect("agent user"); + assert!(db + .set_agent_owner(community, &agent, &human) + .await + .expect("set agent owner")); + db.add_member( + community, + channel_id, + &agent, + buzz_db::channel::MemberRole::Member, + Some(&creator), + ) + .await + .expect("add agent member"); + + let mut workflows = Vec::new(); + for (owner, name, secret) in [ + (&member, "cron-banned-owner", [0u8; 32]), + (&agent, "cron-banned-agent-owner", [1u8; 32]), + (&creator, "cron-control", [2u8; 32]), + ] { + let id = db + .create_workflow( + community, + Some(channel_id), + owner, + name, + &schedule_def(name), + &secret, + ) + .await + .expect("create workflow"); + workflows.push(id); + } + let [banned_wf, agent_wf, control_wf] = + <[Uuid; 3]>::try_from(workflows).expect("three workflows created"); + let engine = Arc::new(WorkflowEngine::new(db.clone(), WorkflowConfig::default())); + let run_count = |workflow_id| { + let db = db.clone(); + async move { + db.list_workflow_runs(community, workflow_id, 10) + .await + .expect("list runs") + .len() + } + }; + + db.ban_community_member(community, &member, &creator, None, None) + .await + .expect("ban owner"); + db.ban_community_member(community, &human, &creator, None, None) + .await + .expect("ban agent's human owner"); + let now = Utc::now(); + engine.cron_tick(now).await; + assert_eq!(run_count(banned_wf).await, 0, "banned owner must not fire"); + assert_eq!( + run_count(agent_wf).await, + 0, + "an agent whose human owner is banned must not fire" + ); + assert_eq!( + run_count(control_wf).await, + 1, + "an unbanned workflow on the same tick still fires" + ); + + for pubkey in [&member, &human] { + assert!(db + .unban_community_member(community, pubkey, &creator) + .await + .expect("unban")); + } + engine.cron_tick(now).await; + assert_eq!( + run_count(banned_wf).await, + 1, + "the denied tick must not have consumed the owner's claim" + ); + assert_eq!( + run_count(agent_wf).await, + 1, + "the denied tick must not have consumed the agent's claim" + ); + assert_eq!( + run_count(control_wf).await, + 1, + "the control's claim for this instant is already taken" + ); + } + + /// The event path fails closed when the owner's restriction state cannot + /// be read, and the same event fires once the read succeeds. + #[tokio::test] + #[ignore = "requires Postgres"] + async fn on_event_denies_run_when_restriction_lookup_fails() { + use sqlx::postgres::PgConnectOptions; + + let db = setup_db().await; + let creator = nostr::Keys::generate().public_key().to_bytes().to_vec(); + let member = nostr::Keys::generate().public_key().to_bytes().to_vec(); + let (community, channel_id) = setup_channel(&db, &creator, &member).await; + let def_json = serde_json::json!({ + "name": "lookup-fails", + "trigger": {"on": "message_posted"}, + "steps": [{"id": "s1", "action": "send_message", "text": "hi"}], + "enabled": true, + }) + .to_string(); + let workflow_id = db + .create_workflow( + community, + Some(channel_id), + &member, + "lookup-fails", + &def_json, + &[0u8; 32], + ) + .await + .expect("create workflow"); + + // Shadow `community_bans` with a view that errors on read; every + // other table still resolves to `public`. + let schema = format!("ban_lookup_fails_{}", Uuid::new_v4().simple()); + sqlx::raw_sql(sqlx::AssertSqlSafe(format!( + "CREATE SCHEMA {schema}; \ + CREATE VIEW {schema}.community_bans AS \ + SELECT * FROM public.community_bans WHERE 1 / (SELECT count(*) * 0 FROM public.communities) = 1;" + ))) + .execute(db.pool()) + .await + .expect("create failing restriction schema"); + let options = test_database_url() + .parse::() + .expect("database url") + .options([("search_path", format!("{schema}, public"))]); + let failing_pool = sqlx::PgPool::connect_with(options) + .await + .expect("failing restriction pool"); + let failing_engine = Arc::new(WorkflowEngine::new( + buzz_db::Db::from_pool(failing_pool.clone()), + WorkflowConfig::default(), + )); + let run_count = || async { + db.list_workflow_runs(community, workflow_id, 10) + .await + .expect("list runs") + .len() + }; + + failing_engine + .on_event(community, &message_event(channel_id)) + .await + .expect("on_event with failing lookup"); + assert_eq!( + run_count().await, + 0, + "a failed restriction lookup must not create a run" + ); + + let engine = Arc::new(WorkflowEngine::new(db.clone(), WorkflowConfig::default())); + engine + .on_event(community, &message_event(channel_id)) + .await + .expect("on_event with healthy lookup"); + assert_eq!(run_count().await, 1, "a readable, unbanned owner fires"); + + failing_pool.close().await; + sqlx::raw_sql(sqlx::AssertSqlSafe(format!("DROP SCHEMA {schema} CASCADE"))) + .execute(db.pool()) + .await + .expect("drop failing restriction schema"); + } } diff --git a/scripts/run-tests.sh b/scripts/run-tests.sh index 07605807bfe..fd1d8a87744 100755 --- a/scripts/run-tests.sh +++ b/scripts/run-tests.sh @@ -115,6 +115,10 @@ run_unit_tests() { run_test_step "buzz-sdk unit tests" \ cargo test -p buzz-sdk --lib -- --nocapture + # buzz-workflow unit tests (infra-free). Mirrors `just test-unit`. + run_test_step "buzz-workflow unit tests" \ + cargo test -p buzz-workflow --lib -- --nocapture + # Keep the relay-to-agent trust-boundary regressions in the fallback path # when cargo-nextest is unavailable. run_test_step "buzz-acp tests" \ From 90e660fce4b37402ce05b13b4ac8721c51766f0e Mon Sep 17 00:00:00 2001 From: Luke Tornquist Date: Thu, 29 Oct 2026 12:03:56 -0400 Subject: [PATCH 09/10] Pin ban-denial WARN expiry through the production cache constructor The logging regression built its own cache and checked only immediate repetition, so dropping the production TTL would have suppressed a workflow's ban WARN until eviction or restart without failing it. The engine and the test now share `ban_denial_warn_cache`, and the test proves WARN, then suppression, then WARN again after the interval. Also records buzz-workflow's new `sqlx` dev-dependency in Cargo.lock. Co-authored-by: Elrond <28d6302a099e5225b02c4155ac4236e4912603df2ab08dbfc2f4fef08ce598c8@buzz.block.builderlab.xyz> Co-Authored-By: Claude Opus 5.5 Signed-off-by: Luke Tornquist --- Cargo.lock | 1 + crates/buzz-workflow/src/lib.rs | 33 +++++++++++++++++++++++++-------- 2 files changed, 26 insertions(+), 8 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index 1cb97df5cdf..c4f1a25cb06 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1615,6 +1615,7 @@ dependencies = [ "serde", "serde_json", "serde_yaml", + "sqlx", "thiserror 2.0.18", "tokio", "tracing", diff --git a/crates/buzz-workflow/src/lib.rs b/crates/buzz-workflow/src/lib.rs index a515af4b8b0..2280045bc8e 100644 --- a/crates/buzz-workflow/src/lib.rs +++ b/crates/buzz-workflow/src/lib.rs @@ -128,10 +128,7 @@ impl WorkflowEngine { .max_capacity(10_000) .time_to_live(std::time::Duration::from_secs(10)) .build(), - ban_denial_warned: moka::sync::Cache::builder() - .max_capacity(10_000) - .time_to_live(BAN_DENIAL_WARN_INTERVAL) - .build(), + ban_denial_warned: ban_denial_warn_cache(BAN_DENIAL_WARN_INTERVAL), } } @@ -798,6 +795,18 @@ impl WorkflowEngine { } } +/// The cache behind [`WorkflowEngine::ban_denial_warned`]: an entry expires +/// `warn_interval` after the WARN that inserted it, so the next ban denial +/// for that workflow warns again. +fn ban_denial_warn_cache( + warn_interval: std::time::Duration, +) -> moka::sync::Cache<(CommunityId, Uuid), ()> { + moka::sync::Cache::builder() + .max_capacity(10_000) + .time_to_live(warn_interval) + .build() +} + /// Whether an owner-authority denial for this workflow logs at WARN. A ban /// denial warns only when no warning for the same workflow is still inside /// `warned`'s TTL; every other denial always warns. @@ -2242,10 +2251,8 @@ steps: #[test] fn ban_denial_warns_once_per_interval_while_other_denials_always_warn() { - let warned = moka::sync::Cache::builder() - .max_capacity(16) - .time_to_live(BAN_DENIAL_WARN_INTERVAL) - .build(); + let interval = std::time::Duration::from_millis(500); + let warned = ban_denial_warn_cache(interval); let community = CommunityId::from_uuid(Uuid::new_v4()); let (wf_a, wf_b) = (Uuid::new_v4(), Uuid::new_v4()); let banned = WorkflowError::OwnerBanned; @@ -2270,6 +2277,16 @@ steps: should_warn_owner_denial(&warned, community, wf_a, &lookup_failed), "a failed lookup is never rate-limited" ); + + std::thread::sleep(interval * 2); + assert!( + should_warn_owner_denial(&warned, community, wf_a, &banned), + "a ban denial warns again once the interval has passed" + ); + assert!( + !should_warn_owner_denial(&warned, community, wf_a, &banned), + "and is suppressed again for the next interval" + ); } fn schedule_def(name: &str) -> String { From 72c8d8bbc09bf7ed5525dabbea0e948c25380cf7 Mon Sep 17 00:00:00 2001 From: Luke Tornquist Date: Thu, 29 Oct 2026 12:07:03 -0400 Subject: [PATCH 10/10] Document the cron tick test's database isolation Co-authored-by: Elrond <28d6302a099e5225b02c4155ac4236e4912603df2ab08dbfc2f4fef08ce598c8@buzz.block.builderlab.xyz> Co-Authored-By: Claude Opus 5.5 Signed-off-by: Luke Tornquist --- crates/buzz-workflow/src/lib.rs | 5 +++++ 1 file changed, 5 insertions(+) diff --git a/crates/buzz-workflow/src/lib.rs b/crates/buzz-workflow/src/lib.rs index 2280045bc8e..36a9bd9ea9b 100644 --- a/crates/buzz-workflow/src/lib.rs +++ b/crates/buzz-workflow/src/lib.rs @@ -2303,6 +2303,11 @@ steps: /// workflow whose human owner is banned, without consuming either /// workflow's scheduled-fire claim: once unbanned, a tick for the same /// instant fires them. An unbanned workflow on the same tick still fires. + /// + /// `cron_tick` scans every enabled workflow in the database. The + /// PostgreSQL lane gives each test process its own database + /// (`scripts/postgres-test-wrapper.sh`), and the assertions read only + /// this test's workflows, so another test's schedules cannot affect it. #[tokio::test] #[ignore = "requires Postgres"] async fn cron_tick_skips_banned_owners_without_consuming_the_claim() {