From 18f97852eefead07ae04af1211eac22bc7124380 Mon Sep 17 00:00:00 2001 From: am <6e30cd56c30e030cd31bb0939b94a7c257c9a09d5ba2d92cf2735da45629f248@buzz.block.builderlab.xyz> Date: Wed, 23 Sep 2026 14:36:56 -0700 Subject: [PATCH] fix: terminate truncated websocket thread windows Make queue-truncation and replacement-topic regression guards falsifiable; cancel finite windows when EVENT or EOSE cannot be queued, and document incomplete-page retry behavior. Co-authored-by: Kalvin Chau Signed-off-by: Kalvin Chau --- .../bridge/thread_window/postgres_tests.rs | 69 ++++++++++++++++++- crates/buzz-relay/src/handlers/req.rs | 14 ++-- crates/buzz-relay/src/protocol.rs | 20 ++++-- docs/nips/NIP-CW.md | 18 +++++ 4 files changed, 110 insertions(+), 11 deletions(-) diff --git a/crates/buzz-relay/src/api/bridge/thread_window/postgres_tests.rs b/crates/buzz-relay/src/api/bridge/thread_window/postgres_tests.rs index 24fb47d1e6..10ada90c9b 100644 --- a/crates/buzz-relay/src/api/bridge/thread_window/postgres_tests.rs +++ b/crates/buzz-relay/src/api/bridge/thread_window/postgres_tests.rs @@ -451,7 +451,7 @@ async fn websocket_batch_enforces_token_scope_and_rejects_queue_truncation() { vec![parsed.clone()], vec![filter.clone()], vec![None], - allowed, + allowed.clone(), f.state.clone(), ) .await; @@ -462,6 +462,51 @@ async fn websocket_batch_enforces_token_scope_and_rejects_queue_truncation() { vec![Some("EVENT"), Some("EVENT"), Some("EOSE")] ); assert_eq!(frames[1][2]["kind"], 39007); + assert!( + !allowed.cancel.is_cancelled(), + "complete page closed socket" + ); + + // Leave room for the two EVENT frames, but not for EOSE. + let (terminal, mut terminal_rx) = f.ws_conn(None); + for _ in 0..2 { + assert!(terminal.send("padding".into())); + } + crate::handlers::req::handle_req( + "terminal".into(), + vec![parsed.clone()], + vec![filter.clone()], + vec![None], + terminal.clone(), + f.state.clone(), + ) + .await; + assert!( + terminal.cancel.is_cancelled(), + "failed EOSE must close socket" + ); + let terminal_frames: Vec<_> = std::iter::from_fn(|| terminal_rx.try_recv().ok()) + .filter_map(|msg| serde_json::from_str::(msg.to_text().unwrap()).ok()) + .collect(); + assert_eq!(terminal_frames.len(), 2); + assert!(terminal_frames.iter().all(|frame| frame[0] == "EVENT")); + + // The full queue also drops an incorrectly attempted EOSE, so checking + // drained frames alone cannot distinguish a complete from a truncated page. + // Assert the production sender reports truncation at the source. + let (probe, _probe_rx) = f.ws_conn(None); + let requests = super::parse(&vec![filter.clone(); 3]).unwrap(); + let sent = crate::handlers::req::serve_thread_windows( + "full", + requests.iter().flatten(), + None, + &f.keys.public_key(), + &probe, + &f.state, + ) + .await + .unwrap(); + assert!(!sent, "a failed EVENT enqueue must not authorize EOSE"); let (slow, mut slow_rx) = f.ws_conn(None); // Capacity four: one row + bounds per filter. Two windows fill it, @@ -471,10 +516,14 @@ async fn websocket_batch_enforces_token_scope_and_rejects_queue_truncation() { vec![parsed.clone(); 3], vec![filter.clone(); 3], vec![None; 3], - slow, + slow.clone(), f.state.clone(), ) .await; + assert!( + slow.cancel.is_cancelled(), + "truncated page must close socket" + ); let drained: Vec<_> = std::iter::from_fn(|| slow_rx.try_recv().ok()) .map(|m| serde_json::from_str::(m.to_text().unwrap()).unwrap()) .collect(); @@ -540,6 +589,14 @@ async fn websocket_window_replaces_live_subscription_before_eose() { .pubsub .retain_topic(&conn.tenant, buzz_pubsub::EventTopic::Channel(f.channel)) .await; + assert_eq!( + f.state + .pubsub + .topic_refcount(&conn.tenant, buzz_pubsub::EventTopic::Channel(f.channel)) + .await, + 1, + "fixture must retain the live channel topic" + ); assert_eq!( f.state .sub_registry @@ -557,6 +614,14 @@ async fn websocket_window_replaces_live_subscription_before_eose() { ) .await; assert!(conn.subscriptions.lock().await.get("replace").is_none()); + assert_eq!( + f.state + .pubsub + .topic_refcount(&conn.tenant, buzz_pubsub::EventTopic::Channel(f.channel)) + .await, + 0, + "replacing the live subscription must release its channel topic" + ); assert!(f .state .sub_registry diff --git a/crates/buzz-relay/src/handlers/req.rs b/crates/buzz-relay/src/handlers/req.rs index f0e6601d2d..a8cc61bddf 100644 --- a/crates/buzz-relay/src/handlers/req.rs +++ b/crates/buzz-relay/src/handlers/req.rs @@ -142,11 +142,17 @@ pub async fn handle_req( .await; match result { Ok(Ok(true)) => { - conn.send(RelayMessage::eose(&sub_id)); + // A window is complete only if its terminal frame was queued. + // A saturated writer must close so the client can retry the + // incomplete page instead of waiting forever for EOSE. + if !conn.send(RelayMessage::eose(&sub_id)) { + conn.cancel.cancel(); + } } Ok(Ok(false)) => { - // A failed enqueue is not a complete page. Do not send EOSE: - // the client must retry rather than accept truncated bounds. + // Some EVENTs could already be queued. Close rather than + // claim a complete page or leave this request silent. + conn.cancel.cancel(); } Ok(Err((status, body))) => close_thread_window_error(&conn, &sub_id, status, &body), Err(_) => { @@ -558,7 +564,7 @@ pub async fn handle_req( ); } -async fn serve_thread_windows<'a>( +pub(crate) async fn serve_thread_windows<'a>( sub_id: &str, requests: impl Iterator, allowed_channels: Option<&[uuid::Uuid]>, diff --git a/crates/buzz-relay/src/protocol.rs b/crates/buzz-relay/src/protocol.rs index 71987a0938..c97c6ce76c 100644 --- a/crates/buzz-relay/src/protocol.rs +++ b/crates/buzz-relay/src/protocol.rs @@ -113,7 +113,7 @@ impl ClientMessage { // Thread-mode cursor validation belongs to the REQ handler, // where invalid input can terminate this subscription with // CLOSED rather than leaving a bare NOTICE. - if value.get("thread_window").is_some_and(|v| v == true) { + if value.get("thread_window").is_some_and(|v| v != false) { return Ok(None); } let Some(raw) = value.get("before_id") else { @@ -352,12 +352,22 @@ mod tests { #[test] fn parse_req_routes_malformed_thread_cursor_to_subscription_validation() { - for cursor in [ - serde_json::json!({"until": 1, "before_id": "zz"}), - serde_json::json!({"before_id": "ab".repeat(32)}), + for (flag, cursor) in [ + ( + serde_json::json!(true), + serde_json::json!({"until": 1, "before_id": "zz"}), + ), + ( + serde_json::json!(true), + serde_json::json!({"before_id": "ab".repeat(32)}), + ), + ( + serde_json::json!("yes"), + serde_json::json!({"before_id": "ab".repeat(32)}), + ), ] { let mut filter = serde_json::json!({ - "thread_window": true, + "thread_window": flag, "#h": [uuid::Uuid::nil()], "#e": ["ab".repeat(32)], "kinds": [9] diff --git a/docs/nips/NIP-CW.md b/docs/nips/NIP-CW.md index d9648e20ea..68af11ece9 100644 --- a/docs/nips/NIP-CW.md +++ b/docs/nips/NIP-CW.md @@ -250,6 +250,14 @@ the head page. Unknown fields, malformed values, and mixing thread windows with another query mode MUST be rejected. A query MAY contain at most four window filters. +On WebSocket REQ, invalid thread-window extensions that reach the window +validator terminate the named subscription with `CLOSED invalid:`. Malformed +standard NIP-01 filter fields (for example `until: -1`, `kinds: [70000]`, or +`limit: "x"`) can fail typed filter deserialization earlier and receive a +connection-level `NOTICE` instead; clients MUST NOT wait indefinitely for EOSE +or CLOSED after a parse failure. A missing or invalid bounds overlay still +makes a page unusable. + ### Relay Processing For an authorized request the relay MUST: @@ -300,6 +308,16 @@ allowance even at `limit: 1`; blind retries of the same query will not help. For timeouts or changed access, retry with bounded backoff and fresh access. These errors MUST NOT trigger legacy compatibility fallback. +A WebSocket thread-window page is complete only after its request-bound bounds +and EOSE arrive. If any EVENT or the terminal EOSE cannot be enqueued, the relay +cancels that connection rather than silently leaving a finite request pending; +a Close frame is best-effort and may not reach a stalled reader. Clients MUST +abandon the entire incomplete batch, retain their previous cursor, and retry +with a fresh request ID and bounded backoff after disconnect or timeout. A client +MUST also impose its own deadline and surface persistent failure rather than +wait forever or fall back to legacy pagination. This relay capability does not +itself enable thread-window pagination in desktop or mobile clients. + ### Thread Bounds: `kind:39007` Bounds are synthesized per query and MUST NOT be accepted at ingest. Tags are