fix(relay): close a WS search whose hit hydration fails instead of sending EOSE (#9148)

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;

crates/buzz-relay/src/handlers/req.rs

@@ -835,7 +835,6 @@ async fn handle_search_req(
     }
 
     let mut seen_ids: HashSet<nostr::EventId> = 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();
     }
 }