From 2eb1fc279cba3bb9fd247887fbbcf1966b0a1c36 Mon Sep 17 00:00:00 2001 From: seiler Date: Mon, 5 Oct 2026 22:45:27 +0000 Subject: [PATCH 1/5] 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 2/5] 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 3/5] 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 4/5] 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 5/5] 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!(