From eb3798c27ad5f3d74eab13199ff88e45ec02cdf3 Mon Sep 17 00:00:00 2001 From: Logan Johnson Date: Fri, 28 Aug 2026 13:59:15 -0400 Subject: [PATCH] Pin PostgreSQL timeout recovery and clear bridge lint Co-authored-by: Larry <8cf5a83f590ec0955b11647d1c88f796a98e088c30a492c58e0e46c3026ae7a4@buzz.block.builderlab.xyz> Signed-off-by: Larry <8cf5a83f590ec0955b11647d1c88f796a98e088c30a492c58e0e46c3026ae7a4@buzz.block.builderlab.xyz> Signed-off-by: Logan Johnson --- crates/buzz-relay/src/api/bridge.rs | 12 +++---- crates/buzz-relay/src/api/workflows.rs | 11 +++++-- .../buzz-relay/src/workflow_delivery_tests.rs | 32 +++++++++++++++++++ 3 files changed, 47 insertions(+), 8 deletions(-) diff --git a/crates/buzz-relay/src/api/bridge.rs b/crates/buzz-relay/src/api/bridge.rs index 998d165ece..05e06e5861 100644 --- a/crates/buzz-relay/src/api/bridge.rs +++ b/crates/buzz-relay/src/api/bridge.rs @@ -1285,7 +1285,7 @@ async fn query_events_authed( // or kind:30622) to a non-owner via the feed path, even though feed SQL // kind allowlists already exclude these kinds. if !crate::handlers::req::event_visible_to_reader( - &state, + state, tenant.community(), &se.event, &pubkey_bytes, @@ -1360,7 +1360,7 @@ async fn query_events_authed( // or kind:30622) to a non-owner via the thread path, even though // requires_h_channel_scope already excludes these kinds from thread metadata. if !crate::handlers::req::event_visible_to_reader( - &state, + state, tenant.community(), &se.event, &pubkey_bytes, @@ -1392,7 +1392,7 @@ async fn query_events_authed( if !seen_aux.insert(se.event.id) || !event_in_accessible_channel(&se, &accessible_channels) || !crate::handlers::req::event_visible_to_reader( - &state, + state, tenant.community(), &se.event, &pubkey_bytes, @@ -1515,7 +1515,7 @@ async fn query_events_authed( // shared-gate (kind:30175 without ["shared","true"]). Single call // covers all three gated event classes. if !crate::handlers::req::event_visible_to_reader( - &state, + state, tenant.community(), &se.event, &pubkey_bytes, @@ -1807,7 +1807,7 @@ async fn count_events_authed( continue; } if !crate::handlers::req::event_visible_to_reader( - &state, + state, tenant.community(), &se.event, &pubkey_bytes, @@ -1881,7 +1881,7 @@ async fn count_events_authed( continue; } if !crate::handlers::req::event_visible_to_reader( - &state, + state, tenant.community(), &se.event, &pubkey_bytes, diff --git a/crates/buzz-relay/src/api/workflows.rs b/crates/buzz-relay/src/api/workflows.rs index 657e65306f..1bd1f371a5 100644 --- a/crates/buzz-relay/src/api/workflows.rs +++ b/crates/buzz-relay/src/api/workflows.rs @@ -268,7 +268,7 @@ fn approval_json(approval: &buzz_db::workflow::ApprovalRecord) -> Value { // A missing/revoked authority is terminal; unavailable storage is not. Keep // this distinction at the endpoint so ACP's bounded transport retries can work. -fn wake_lookup_error(error: buzz_db::DbError) -> (StatusCode, Json) { +pub(crate) fn wake_lookup_error(error: buzz_db::DbError) -> (StatusCode, Json) { use buzz_db::DbError; let status = match &error { DbError::NotFound(_) => StatusCode::NOT_FOUND, @@ -280,7 +280,14 @@ fn wake_lookup_error(error: buzz_db::DbError) -> (StatusCode, Json) { code.starts_with("08") || matches!( code.as_ref(), - "40001" | "40P01" | "53300" | "57P01" | "57P02" | "57P03" + "40001" + | "40P01" + | "53300" + | "55P03" + | "57014" + | "57P01" + | "57P02" + | "57P03" ) }) => { diff --git a/crates/buzz-relay/src/workflow_delivery_tests.rs b/crates/buzz-relay/src/workflow_delivery_tests.rs index 2407e47839..3b9c0dd0d8 100644 --- a/crates/buzz-relay/src/workflow_delivery_tests.rs +++ b/crates/buzz-relay/src/workflow_delivery_tests.rs @@ -488,3 +488,35 @@ async fn notification_failure_rolls_back_message_mentions_and_thread_metadata() .is_some()); } } + +#[tokio::test] +#[ignore = "requires Postgres and Redis"] +async fn storage_timeout_is_retryable_but_missing_authority_is_terminal() { + use crate::api::workflows::wake_lookup_error; + let f = Fixture::new().await; + let mut tx = f.state.db.begin_transaction().await.expect("transaction"); + sqlx::query("SET LOCAL statement_timeout = '10ms'") + .execute(&mut *tx) + .await + .expect("set timeout"); + let error = sqlx::query("SELECT pg_sleep(1)") + .execute(&mut *tx) + .await + .expect_err("statement timeout"); + assert_eq!( + error.as_database_error().and_then(|e| e.code()).as_deref(), + Some("57014") + ); + assert_eq!( + wake_lookup_error(error.into()).0, + StatusCode::SERVICE_UNAVAILABLE + ); + tx.rollback().await.expect("rollback"); + let error = f + .state + .db + .get_workflow_run(f.community, Uuid::new_v4()) + .await + .expect_err("missing run"); + assert_eq!(wake_lookup_error(error).0, StatusCode::NOT_FOUND); +}