diff --git a/crates/buzz-relay/src/api/bridge.rs b/crates/buzz-relay/src/api/bridge.rs index 131ffef48c..0db0a7e125 100644 --- a/crates/buzz-relay/src/api/bridge.rs +++ b/crates/buzz-relay/src/api/bridge.rs @@ -10573,10 +10573,14 @@ pub(crate) mod postgres_tests { /// A shed history read answers 503 `database unavailable` on every /// bridge `/query` branch that serves one: channel window, the three - /// feeds and legacy thread replies. A saturated reader (every connection - /// checked out, none idle) rejects history reads instead of moving them - /// to the writer, the same `PoolTimedOut` as a full history gate. - /// Mutation: map any of these branches through `internal_error` → RED. + /// feeds, legacy thread replies and search hydration. A WS search whose + /// hydration is shed closes with the retryable `error: database error` + /// instead of an EOSE that would read as "no matches". A saturated reader + /// (every connection checked out, none idle) rejects history reads + /// instead of moving them to the writer, the same `PoolTimedOut` as a + /// full history gate. Mutations: map any of these branches through + /// `internal_error`, or let a failed WS hydration fall through to EOSE + /// → RED. #[tokio::test] #[ignore = "requires Postgres"] async fn shed_history_read_maps_to_503_on_every_bridge_branch() { @@ -10603,11 +10607,14 @@ pub(crate) mod postgres_tests { "thread replies", serde_json::json!({"kinds": [9], "#e": [&fx.root], "depth_limit": 1}), ), + ("search hydrate", fx.search_filter()), ]; for (branch, filter) in &cases { let (status, body) = fx.http("/query", filter).await; assert_eq!(status, StatusCode::OK, "healthy {branch}: {body}"); } + let frame = fx.ws_search().await; + assert!(frame.contains("root needle"), "healthy WS search: {frame}"); let mut held = Vec::new(); for _ in 0..fx.replica.options().get_max_connections() { @@ -10624,6 +10631,14 @@ pub(crate) mod postgres_tests { "shed {branch}: {body}" ); } + assert_eq!( + fx.ws_search().await, + format!( + r#"["CLOSED","s","{}"]"#, + crate::handlers::req::DATABASE_ERROR_CLOSED + ), + "shed WS search hydration" + ); drop(held); fx.drop().await; diff --git a/crates/buzz-relay/src/handlers/req.rs b/crates/buzz-relay/src/handlers/req.rs index fe468c6b22..989c2957a2 100644 --- a/crates/buzz-relay/src/handlers/req.rs +++ b/crates/buzz-relay/src/handlers/req.rs @@ -835,7 +835,6 @@ async fn handle_search_req( } let mut seen_ids: HashSet = HashSet::new(); - let mut failure = None; for filter in filters { let search_text = match &filter.search { @@ -981,13 +980,25 @@ async fn handle_search_req( return; } Err(e) => { + // CLOSED, not EOSE: an EOSE after a failed hydration + // would mark the hits never fetched as "no matches". + // That includes a history read shed while the reader + // is saturated, which the client should retry. warn!( sub_id = %sub_id, error_kind = %e.kind(), "NIP-50 batch fetch failed" ); - failure = Some("database_error"); - break; + access.error("database_error"); + super::close::close_if_owner( + sub_id, + owner, + Some(DATABASE_ERROR_CLOSED), + conn, + state, + ) + .await; + return; } }; @@ -1080,11 +1091,8 @@ async fn handle_search_req( } } - let queued = super::close::send_if_owner(sub_id, owner, RelayMessage::eose(sub_id), conn).await; - match failure { - Some(reason) => access.error(reason), - None if queued => access.served(), - None => {} + if super::close::send_if_owner(sub_id, owner, RelayMessage::eose(sub_id), conn).await { + access.served(); } }