From b58cb75cfe8ffce6aeb2a134e7b8c78b76b15c50 Mon Sep 17 00:00:00 2001 From: tomdebone Date: Thu, 10 Sep 2026 20:30:20 +0200 Subject: [PATCH] =?UTF-8?q?fix(at-firehose):=20Jetstream-Filter=20wirkte?= =?UTF-8?q?=20nie=20=E2=80=94=20Collections=20in=20die=20URL?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Der Consumer verband sich auf die nackte URL und schickte danach `{"type":"options","wantedCollections":[…]}` als Textframe. Jetstream ignoriert das, und zwar stillschweigend: Filter sind Query-Parameter, und der einzige nachrichtenbasierte Weg (`options_update`) verlangt, dass die Verbindung mit `requireHello=true` geöffnet wurde. Jede Instanz, die glaubte, sechs Collections zu abonnieren, hat also den kompletten öffentlichen Firehose gezogen. Gemessen gegen jetstream1.us-east: 3119 Events in 8 s ungefiltert, 520 für eine einzelne Collection, 12 für die beiden, die dieses Projekt wirklich braucht. Konkrete Folgen: die Dev-Datenbank ist unbemerkt auf 3,3 Mio. Posts gewachsen, und auf der Produktionsinstanz musste die AppView abgeschaltet und aus dem Autostart genommen werden, weil sie die Platte vollzuschreiben drohte — dort stand `JETSTREAM_COLLECTIONS=app.twi.post` korrekt in der .env und wurde einfach nicht beachtet. Was der Fix NICHT löst, und das steht auch so im Code: die Collections, die ein Bluesky-artiges Produkt normalerweise will (post/like/repost/follow), sind ~97 % des Volumens. Richtig zu filtern ist notwendig, nicht hinreichend. Nebenbei: upload_blob_rejects_oversized fiel etwa jeden dritten Lauf um. Der Server bricht die Verbindung ab, sobald das Body-Limit reißt, also sieht der Client je nach Timing die 413 oder einen Reset beim Schreiben. Beides beweist, dass der Upload abgelehnt wurde; der Test akzeptiert jetzt beides. Co-Authored-By: Claude Opus 5 Claude-Session: https://claude.ai/code/session_013HC9HLrUU1LNwkzp8nkDLX --- crates/at-firehose/src/consumer.rs | 114 +++++++++++++++++--- crates/pds-server/tests/blob_integration.rs | 27 +++-- 2 files changed, 119 insertions(+), 22 deletions(-) diff --git a/crates/at-firehose/src/consumer.rs b/crates/at-firehose/src/consumer.rs index 8017c90..a4e6ac2 100644 --- a/crates/at-firehose/src/consumer.rs +++ b/crates/at-firehose/src/consumer.rs @@ -1,6 +1,5 @@ use anyhow::Result; -use futures::{SinkExt, StreamExt}; -use serde_json::json; +use futures::StreamExt; use std::sync::Arc; use std::sync::atomic::{AtomicBool, Ordering}; use std::time::Duration; @@ -33,6 +32,43 @@ impl JetstreamConsumer { } } + /// The URL actually dialled: base URL plus `wantedCollections` and + /// `cursor` as query parameters. + /// + /// This used to connect to the bare URL and then send + /// `{"type": "options", "wantedCollections": [...]}` as a text frame. + /// Jetstream ignores that, and silently: filters are query parameters, + /// and the only message-based path (`options_update`) requires the + /// connection to have been opened with `requireHello=true`. So every + /// deployment that thought it was subscribing to six collections was in + /// fact taking the entire public firehose — measured against + /// jetstream1.us-east: 3119 events in 8 s unfiltered versus 520 for a + /// single collection. On the dev database that quietly grew to 3.3 M + /// posts; on the production instance the AppView had to be switched off + /// to stop it filling the disk. + /// + /// Note what fixing this does *not* solve: the collections this project + /// wants (`app.bsky.feed.post` / `like` / `repost` / `graph.follow`) are + /// ~97 % of the firehose by volume. Correct filtering is necessary, not + /// sufficient — an instance that does not want the whole public network + /// in its index wants `wantedDids`, or no Jetstream at all. + pub fn subscribe_url(&self) -> String { + let mut url = self.url.trim_end_matches('&').to_string(); + let mut sep = if url.contains('?') { '&' } else { '?' }; + for c in &self.collections { + url.push(sep); + url.push_str("wantedCollections="); + url.push_str(&urlencode(c)); + sep = '&'; + } + if self.cursor_us > 0 { + url.push(sep); + url.push_str("cursor="); + url.push_str(&self.cursor_us.to_string()); + } + url + } + /// Build a consumer that shares a connection-state flag with the caller. pub fn with_connected_flag(mut self, flag: Arc) -> Self { self.connected = Some(flag); @@ -81,23 +117,13 @@ impl JetstreamConsumer { F: FnMut(JetstreamEvent) -> Fut + Send, Fut: std::future::Future> + Send, { - let (mut ws, _) = tokio_tungstenite::connect_async(&self.url).await?; - info!("connected to jetstream: {}", self.url); + let url = self.subscribe_url(); + let (mut ws, _) = tokio_tungstenite::connect_async(&url).await?; + info!("connected to jetstream: {url}"); if let Some(flag) = &self.connected { flag.store(true, Ordering::Relaxed); } - if !self.collections.is_empty() || self.cursor_us > 0 { - let mut options = json!({ "type": "options" }); - if !self.collections.is_empty() { - options["wantedCollections"] = json!(self.collections); - } - if self.cursor_us > 0 { - options["cursor"] = json!(self.cursor_us); - } - ws.send(Message::Text(options.to_string())).await?; - } - while let Some(msg) = ws.next().await { let msg = msg?; if let Message::Text(text) = msg { @@ -109,3 +135,61 @@ impl JetstreamConsumer { Ok(()) } } + +/// Percent-encode everything outside the unreserved set. Collection NSIDs are +/// dots and letters today, but a `wantedDids` value carries `:` — encoding +/// unconditionally keeps this correct if the caller passes one. +fn urlencode(s: &str) -> String { + s.chars() + .map(|c| match c { + 'A'..='Z' | 'a'..='z' | '0'..='9' | '-' | '_' | '.' | '~' => c.to_string(), + other => { + let mut buf = [0u8; 4]; + other + .encode_utf8(&mut buf) + .bytes() + .map(|b| format!("%{b:02X}")) + .collect() + } + }) + .collect() +} + +#[cfg(test)] +mod url_tests { + use super::*; + + #[test] + fn collections_go_into_the_query_string() { + let c = JetstreamConsumer::new( + "wss://jetstream1.us-east.bsky.network/subscribe", + vec!["app.twi.post".into(), "app.bsky.feed.like".into()], + ); + assert_eq!( + c.subscribe_url(), + "wss://jetstream1.us-east.bsky.network/subscribe\ + ?wantedCollections=app.twi.post&wantedCollections=app.bsky.feed.like" + .replace(' ', "") + ); + } + + #[test] + fn cursor_is_appended_and_respects_an_existing_query() { + let mut c = JetstreamConsumer::new("wss://host/subscribe?compress=false", vec![]); + c.cursor_us = 1234; + assert_eq!(c.subscribe_url(), "wss://host/subscribe?compress=false&cursor=1234"); + } + + #[test] + fn no_filters_leaves_the_url_alone() { + let c = JetstreamConsumer::new("wss://host/subscribe", vec![]); + assert_eq!(c.subscribe_url(), "wss://host/subscribe"); + } + + /// A DID contains `:`, which has to survive as `%3A` in a query value. + #[test] + fn values_are_percent_encoded() { + assert_eq!(urlencode("did:plc:abc"), "did%3Aplc%3Aabc"); + assert_eq!(urlencode("app.bsky.feed.post"), "app.bsky.feed.post"); + } +} diff --git a/crates/pds-server/tests/blob_integration.rs b/crates/pds-server/tests/blob_integration.rs index bff6921..6d60e9a 100644 --- a/crates/pds-server/tests/blob_integration.rs +++ b/crates/pds-server/tests/blob_integration.rs @@ -314,19 +314,32 @@ async fn upload_blob_rejects_oversized() { // 413 from axum's body extractor. let payload = vec![0u8; 2 * 1024 * 1024]; - let resp = c + // Two legitimate outcomes, and which one happens is a race the test + // cannot win: the limit trips while the client is still writing the + // 2 MiB body. If the rejection reaches the socket first, the client + // reads `413`; if the server closes its side first, the client's + // write fails with a connection reset and never gets to read a + // status. Asserting only on `413` made this test fail roughly one run + // in three. What actually matters — and what both outcomes prove — is + // that the upload was refused rather than accepted. + match c .post(format!("{}/xrpc/com.atproto.uploadBlob", PDS_URL)) .bearer_auth(&jwt) .header("Content-Type", "image/png") .body(payload) .send() .await - .unwrap(); - assert_eq!( - resp.status().as_u16(), - 413, - "oversized upload must return 413" - ); + { + Ok(resp) => assert_eq!( + resp.status().as_u16(), + 413, + "oversized upload must be refused with 413" + ), + Err(e) => assert!( + e.is_request(), + "the only acceptable error is the server hanging up mid-body, got {e:?}" + ), + } } /// `com.atproto.uploadBlob` rejects requests with no `Authorization`