jerry/buzz
1
0
Fork 0
mirror of https://github.com/block/buzz.git synced 2026-10-28 21:11:48 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix: terminate truncated websocket thread windows

Browse Source 
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 <kalvin@block.xyz>

Signed-off-by: Kalvin Chau <kalvin@block.xyz>
This commit is contained in:
am
2026-09-23 14:36:56 -07:00
parent b082ae71f8
commit 18f97852ee
4 changed files with 110 additions and 11 deletions
Download Patch File Download Diff File Expand all files Collapse all files

69
crates/buzz-relay/src/api/bridge/thread_window/postgres_tests.rs
View File

@@ -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::<Value>(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::<Value>(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

14
crates/buzz-relay/src/handlers/req.rs
View File

@@ -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<Item = &'a buzz_core::thread_window::Request>,
allowed_channels: Option<&[uuid::Uuid]>,

20
crates/buzz-relay/src/protocol.rs
View File

@@ -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]

18
docs/nips/NIP-CW.md
View File

@@ -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