From 745df712f3e347e3b89e0f79f690696b2d7fe558 Mon Sep 17 00:00:00 2001 From: dawn <90008@gaze.systems> Date: Wed, 27 May 2026 16:21:41 +0000 Subject: [PATCH] [jetstream] stage backfill commits also --- src/ops.rs | 1 + src/types.rs | 18 ++++++++++++++++++ tests/stream_jetstream_subscribe.nu | 59 +++++++++++++++++++++++++++++++++++++++++++++++++++++++++-- docs/api/jetstream.md | 12 +++++++----- src/api/jetstream.rs | 19 +++++++++++++++++-- src/backfill/mod.rs | 22 ++++++++++++++++++++++ src/control/jetstream.rs | 68 ++++++++++++++++++++++++++++++++++++++++++++++++++++++++------------ src/control/stream.rs | 5 ++++- src/db/mod.rs | 4 ++-- src/ingest/relay.rs | 8 ++++---- 10 file(s) changed, 188 insertion(s)(+), 28 deletion(s)(-) diff --git a/src/ops.rs b/src/ops.rs --- a/src/ops.rs +++ b/src/ops.rs @@ -384,6 +384,7 @@ did: did_trimmed.clone().into_static(), collection: collection.clone().into_static(), event_id, + live: true, }; jetstream_events.push(crate::jetstream::stage_event(batch, db, jetstream)?); } diff --git a/src/types.rs b/src/types.rs --- a/src/types.rs +++ b/src/types.rs @@ -436,6 +436,12 @@ pub inline_block: Option, } +#[inline] +#[allow(dead_code)] +pub(crate) fn default_true() -> bool { + true +} + #[cfg(feature = "jetstream")] #[derive(Debug, Serialize, Deserialize, Clone)] #[serde(bound(deserialize = "'i: 'de"))] @@ -447,6 +453,8 @@ #[serde(borrow)] collection: CowStr<'i>, event_id: u64, + #[serde(default = "crate::types::default_true")] + live: bool, }, #[cfg(feature = "relay")] RelayCommit { @@ -498,6 +506,14 @@ #[cfg(feature = "jetstream")] impl<'i> StoredJetstreamEvent<'i> { + pub(crate) fn is_live(&self) -> bool { + match self { + #[cfg(feature = "indexer_stream")] + Self::Commit { live, .. } => *live, + _ => true, + } + } + pub(crate) fn did(&self) -> &TrimmedDid<'i> { match self { #[cfg(feature = "indexer_stream")] @@ -529,10 +545,12 @@ did, collection, event_id, + live, } => StoredJetstreamEvent::Commit { did: did.into_static(), collection: collection.into_static(), event_id, + live, }, #[cfg(feature = "relay")] Self::RelayCommit { diff --git a/tests/stream_jetstream_subscribe.nu b/tests/stream_jetstream_subscribe.nu --- a/tests/stream_jetstream_subscribe.nu +++ b/tests/stream_jetstream_subscribe.nu @@ -6,7 +6,7 @@ # # Jetstream event format: # {did, time_us, kind: "commit"|"identity"|"account", commit?|identity?|account?} -# commit: {rev, operation, collection, rkey, record?, cid?} +# commit: {rev, operation, collection, rkey, record?, cid?, live} # identity: {did, seq, time, handle?} # account: {did, active, seq, time, status?} @@ -43,7 +43,7 @@ } def assert-commit-structure [c: record, label: string, ...pids: int] { - for field in ["rev", "operation", "collection", "rkey"] { + for field in ["rev", "operation", "collection", "rkey", "live"] { if not ($field in $c) { fail $"($label): commit missing field ($field)" ...$pids } @@ -109,6 +109,25 @@ if not (wait-for-backfill $url) { fail "backfill timed out" $instance.pid } + + # --- verify historical commits exist after backfill --- + print "--- verifying historical commits exist after backfill ---" + let hist_verify_file = $"($db_path)/hist_verify.txt" + # use cursor=0 to replay everything; we expect historical commits for our DID + let hist_verify_events = (collect-ws-json $"($ws_base)/subscribe?cursor=0" $hist_verify_file 10sec) + let hist_verify_commits = ($hist_verify_events | where { |e| + ($e | get -o kind | default "") == "commit" and ($e | get -o did | default "") == $did + }) + print $"backfill historical commits for our DID: ($hist_verify_commits | length)" + if ($hist_verify_commits | is-empty) { + fail "expected historical commits for our DID after backfill, got none" $instance.pid + } + # all backfill commits must have live: false + let any_live_hist = ($hist_verify_commits | any { |e| ($e | get -o commit.live | default true) == true }) + if $any_live_hist { + fail "backfill commits must have live=false" $instance.pid + } + print "historical commits verified (live=false)" # start live subscriber before creating records so we catch the events let live_file = $"($db_path)/live.txt" @@ -255,6 +274,42 @@ fail "wantedDids filter returned events for wrong DID" $instance.pid } print "wantedDids filter is correct" + + # --- scenario: wantedEventTypes filter --- + print "--- scenario: wantedEventTypes filter ---" + let type_live_file = $"($db_path)/type_live_filter.txt" + let type_live_url = $"($ws_base)/subscribe?cursor=($cursor_us)&wantedEventTypes=live" + let type_live_events = (collect-ws-json $type_live_url $type_live_file 8sec) + print $"wantedEventTypes=live: ($type_live_events | length) events" + + let type_hist_file = $"($db_path)/type_hist_filter.txt" + let type_hist_url = $"($ws_base)/subscribe?cursor=($cursor_us)&wantedEventTypes=historical" + let type_hist_events = (collect-ws-json $type_hist_url $type_hist_file 8sec) + print $"wantedEventTypes=historical: ($type_hist_events | length) events" + + assert-no-error-events $type_live_events "wantedEventTypes=live filter" $instance.pid + assert-no-error-events $type_hist_events "wantedEventTypes=historical filter" $instance.pid + + # verify live filter only returns live commits + let live_commits = ($type_live_events | where { |e| ($e | get -o kind | default "") == "commit" and ($e | get -o did | default "") == $did }) + let any_non_live = ($live_commits | any { |e| ($e | get -o commit.live | default false) == false }) + if $any_non_live { + fail "wantedEventTypes=live returned non-live commits" $instance.pid + } + print $"wantedEventTypes=live: ($live_commits | length) live commits for our DID" + + # verify historical filter returns historical commits for our DID + let hist_commits = ($type_hist_events | where { |e| ($e | get -o kind | default "") == "commit" and ($e | get -o did | default "") == $did }) + if ($hist_commits | is-empty) { + fail "wantedEventTypes=historical: expected historical commits for our DID, got none" $instance.pid + } + let any_live_hist = ($hist_commits | any { |e| ($e | get -o commit.live | default true) == true }) + if $any_live_hist { + fail "wantedEventTypes=historical returned live commits" $instance.pid + } + print $"wantedEventTypes=historical: ($hist_commits | length) historical commits for our DID, all have live=false" + + print "wantedEventTypes filter is correct" try { kill $instance.pid } print "=== jetstream /subscribe test PASSED ===" diff --git a/docs/api/jetstream.md b/docs/api/jetstream.md --- a/docs/api/jetstream.md +++ b/docs/api/jetstream.md @@ -12,8 +12,9 @@ | param | type | description | | :--- | :--- | :--- | -| `wantedCollections` | string \| seq | list of collection NSIDs to receive (e.g. `app.bsky.feed.post`). supports namespace wildcards (e.g. `app.bsky.feed.*`). | -| `wantedDids` | string \| seq | list of DIDs to receive (e.g. `did:plc:abc123xyz`). | +| `wantedCollections` | seq | list of collection NSIDs to receive (e.g. `app.bsky.feed.post`). supports namespace wildcards (e.g. `app.bsky.feed.*`). | +| `wantedDids` | seq | list of DIDs to receive (e.g. `did:plc:abc123xyz`). | +| `wantedEventTypes` | seq | list of event types to receive: `live` or `historical`. if not specified, both are returned. | | `maxMessageSizeBytes` | integer | filters out events whose serialized JSON size exceeds this value. | | `cursor` | integer | unix microseconds timestamp (`time_us`) to replay historical events from. | | `compress` | boolean | if `true` (or if header `Socket-Encoding` contains `zstd`), compresses frames using zstd and sends them as binary websocket frames. | @@ -21,7 +22,7 @@ ### in-stream options update -clients can dynamically modify filtering criteria (`wantedCollections`, `wantedDids`, and `maxMessageSizeBytes`) without reconnecting by sending a text frame with the following JSON format: +clients can dynamically modify filtering criteria (`wantedCollections`, `wantedDids`, `maxMessageSizeBytes`, and `wantedEventTypes`) without reconnecting by sending a text frame with the following JSON format: ```json { @@ -29,7 +30,8 @@ "payload": { "wantedCollections": ["app.bsky.feed.post", "app.bsky.like.*"], "wantedDids": ["did:plc:abc123xyz"], - "maxMessageSizeBytes": 5000000 + "maxMessageSizeBytes": 5000000, + "wantedEventTypes": ["live"] } } ``` @@ -106,5 +108,5 @@ ### additional details -- **live stream scope**: jetstream subscribers only receive live firehose events. historical backfill and sync replay events are processed with `live: false` internally, and are never staged in the jetstream keyspace or broadcasted. +- **live stream scope**: jetstream subscribers receive both live firehose events (`live: true`) and historical backfill / sync replay events (`live: false`) by default. be aware that historical backfills will interleave with live events unless filtered using the `wantedEventTypes` query parameter. - **slow consumers**: if a socket's send buffer remains full for longer than the configured timeout, the server sends a `{"type":"error","error":"ConsumerTooSlow"}` message and drops the connection. diff --git a/src/api/jetstream.rs b/src/api/jetstream.rs --- a/src/api/jetstream.rs +++ b/src/api/jetstream.rs @@ -1,4 +1,3 @@ -use std::sync::Arc; use std::time::Duration; use axum::extract::{Query, State}; @@ -72,6 +71,12 @@ compress: bool, #[serde(default, rename = "requireHello")] require_hello: bool, + #[serde( + default, + rename = "wantedEventTypes", + deserialize_with = "deserialize_string_or_seq" + )] + wanted_event_types: Vec, } pub async fn handle_subscribe( @@ -84,6 +89,7 @@ &query.wanted_collections, &query.wanted_dids, parse_max_message_size(query.max_message_size_bytes), + &query.wanted_event_types, ) { Ok(options) => JetstreamFilter::new(options), Err(err) => return (StatusCode::BAD_REQUEST, err).into_response(), @@ -241,6 +247,7 @@ &payload.wanted_collections, &payload.wanted_dids, parse_max_message_size(Some(payload.max_message_size_bytes)), + &payload.wanted_event_types, )?; options.update(next); Ok(true) @@ -275,8 +282,14 @@ wanted_collections: &[String], wanted_dids: &[String], max_message_size_bytes: u32, + wanted_event_types: &[String], ) -> Result { - JetstreamSubscriberOptions::parse(wanted_collections, wanted_dids, max_message_size_bytes) + JetstreamSubscriberOptions::parse( + wanted_collections, + wanted_dids, + max_message_size_bytes, + wanted_event_types, + ) } fn parse_max_message_size(value: Option) -> u32 { @@ -312,4 +325,6 @@ wanted_dids: Vec, #[serde(default, rename = "maxMessageSizeBytes", alias = "maxSize")] max_message_size_bytes: i64, + #[serde(default, rename = "wantedEventTypes")] + wanted_event_types: Vec, } diff --git a/src/backfill/mod.rs b/src/backfill/mod.rs --- a/src/backfill/mod.rs +++ b/src/backfill/mod.rs @@ -702,6 +702,17 @@ }; let bytes = rmp_serde::to_vec(&evt).into_diagnostic()?; batch.insert(&app_state.db.events, keys::event_key(event_id), bytes); + + #[cfg(feature = "jetstream")] + { + let jetstream = crate::types::StoredJetstreamEvent::Commit { + did: TrimmedDid::from(&did).into_static(), + collection: CowStr::Borrowed(collection).into_static(), + event_id, + live: false, + }; + crate::jetstream::stage_event(&mut batch, &app_state.db, jetstream)?; + } } count += 1; @@ -745,6 +756,17 @@ }; let bytes = rmp_serde::to_vec(&evt).into_diagnostic()?; batch.insert(&app_state.db.events, keys::event_key(event_id), bytes); + + #[cfg(feature = "jetstream")] + { + let jetstream = crate::types::StoredJetstreamEvent::Commit { + did: TrimmedDid::from(&did).into_static(), + collection: CowStr::Borrowed(&collection).into_static(), + event_id, + live: false, + }; + crate::jetstream::stage_event(&mut batch, &app_state.db, jetstream)?; + } } delta -= 1; diff --git a/src/control/jetstream.rs b/src/control/jetstream.rs --- a/src/control/jetstream.rs +++ b/src/control/jetstream.rs @@ -62,6 +62,7 @@ wanted_collections: Option, wanted_dids: Arc>>, max_message_size_bytes: u32, + wanted_event_types: Option, } #[derive(Clone)] @@ -70,18 +71,27 @@ full_paths: HashSet, } +#[derive(Clone)] +struct WantedEventTypes { + live: bool, + historical: bool, +} + impl JetstreamSubscriberOptions { pub fn parse( wanted_collections: &[String], wanted_dids: &[String], max_message_size_bytes: u32, + wanted_event_types: &[String], ) -> std::result::Result { let wanted_collections = parse_wanted_collections(wanted_collections)?; let wanted_dids = parse_wanted_dids(wanted_dids)?; + let wanted_event_types = parse_wanted_event_types(wanted_event_types)?; Ok(Self { wanted_collections, wanted_dids: Arc::new(wanted_dids), max_message_size_bytes, + wanted_event_types, }) } @@ -90,6 +100,16 @@ } pub(crate) fn wants(&self, event: &StoredJetstreamEvent<'_>) -> bool { + if let Some(wanted) = &self.wanted_event_types { + let is_live = event.is_live(); + if is_live && !wanted.live { + return false; + } + if !is_live && !wanted.historical { + return false; + } + } + if !self.wanted_dids.is_empty() { let mut did = Vec::with_capacity(event.did().len()); event.did().write_to_vec(&mut did); @@ -180,6 +200,24 @@ .map_err(|e| e.to_string()) } +fn parse_wanted_event_types( + provided: &[String], +) -> std::result::Result, String> { + if provided.is_empty() { + return Ok(None); + } + let mut live = false; + let mut historical = false; + for ev in provided { + match ev.as_str() { + "live" => live = true, + "historical" => historical = true, + _ => return Err(format!("unknown wantedEventTypes value: {}", ev)), + } + } + Ok(Some(WantedEventTypes { live, historical })) +} + fn parse_wanted_dids(provided: &[String]) -> std::result::Result>, String> { let mut wanted = HashSet::new(); for did_raw in provided { @@ -206,40 +244,46 @@ #[test] fn collection_prefixes_accept_domain_prefixes() { - let opts = JetstreamSubscriberOptions::parse(&["app.bsky.*".into()], &[], 0).unwrap(); + let opts = JetstreamSubscriberOptions::parse(&["app.bsky.*".into()], &[], 0, &[]).unwrap(); let wanted = opts.wanted_collections.unwrap(); assert!(wanted.prefixes.iter().any(|prefix| prefix == "app.bsky.")); - assert!(JetstreamSubscriberOptions::parse(&["app.bsky.feed.po*".into()], &[], 0).is_err()); + assert!(JetstreamSubscriberOptions::parse(&["app.bsky.feed.po*".into()], &[], 0, &[]).is_err()); } #[test] - fn full_path_filter_matches_relay_commit_collection() { + fn full_path_filter_matches_commit_collection() { use crate::db::types::TrimmedDid; use crate::types::StoredJetstreamEvent; use jacquard_common::{CowStr, IntoStatic}; use smol_str::ToSmolStr; - let opts = JetstreamSubscriberOptions::parse(&["app.bsky.feed.post".into()], &[], 0) + let opts = JetstreamSubscriberOptions::parse(&["app.bsky.feed.post".into()], &[], 0, &[]) .expect("app.bsky.feed.post must be a valid collection"); let did = TrimmedDid::from(&jacquard_common::types::string::Did::new("did:plc:abc123").unwrap()) .into_static(); - let matching = StoredJetstreamEvent::RelayCommit { + let matching = StoredJetstreamEvent::Commit { did: did.clone(), collection: CowStr::Owned("app.bsky.feed.post".to_smolstr()), - relay_seq: 1, - op_index: 0, + event_id: 1, + live: true, }; - let non_matching = StoredJetstreamEvent::RelayCommit { + let non_matching = StoredJetstreamEvent::Commit { did: did.clone(), collection: CowStr::Owned("app.bsky.actor.profile".to_smolstr()), - relay_seq: 2, - op_index: 0, + event_id: 2, + live: true, }; - let account = StoredJetstreamEvent::RelayAccount { did, relay_seq: 3 }; + let account = StoredJetstreamEvent::Account { + did, + active: true, + status: None, + seq: 1, + time: crate::ingest::stream::Datetime(chrono::Utc::now().into()), + }; assert!( opts.wants(&matching), @@ -259,7 +303,7 @@ fn wanted_did_limit_counts_unique_dids() { let dids = vec!["did:plc:abc123".to_string(); 10_001]; - let opts = JetstreamSubscriberOptions::parse(&[], &dids, 0).unwrap(); + let opts = JetstreamSubscriberOptions::parse(&[], &dids, 0, &[]).unwrap(); assert_eq!(opts.wanted_dids.len(), 1); } diff --git a/src/control/stream.rs b/src/control/stream.rs --- a/src/control/stream.rs +++ b/src/control/stream.rs @@ -940,6 +940,7 @@ record: Option<&'a serde_json::Value>, #[serde(skip_serializing_if = "Option::is_none")] cid: Option, + live: bool, } #[cfg(feature = "jetstream")] @@ -975,7 +976,7 @@ match &event.event { #[cfg(feature = "indexer_stream")] - StoredJetstreamEvent::Commit { event_id, .. } => { + StoredJetstreamEvent::Commit { event_id, live, .. } => { let bytes = state.db.events.get(keys::event_key(*event_id)).ok()??; let stored: StoredEvent = rmp_serde::from_slice(&bytes).ok()?; let evt = stored_to_event(state, *event_id, stored, None)?; @@ -993,6 +994,7 @@ rkey: rec.rkey.as_str(), record: rec.record.as_ref(), cid: rec.cid.map(|cid| cid.to_string()), + live: *live, }, }, }; @@ -1044,6 +1046,7 @@ rkey, record: record_owned.as_ref(), cid, + live: true, }, }, }; diff --git a/src/db/mod.rs b/src/db/mod.rs --- a/src/db/mod.rs +++ b/src/db/mod.rs @@ -78,7 +78,7 @@ pub(crate) next_jetstream_id: Arc, #[cfg(feature = "jetstream")] pub(crate) last_jetstream_time_us: Arc, - #[cfg(feature = "jetstream")] + #[cfg(all(feature = "jetstream", feature = "relay"))] pub(crate) jetstream_lock: Arc>, #[cfg(feature = "relay")] pub(crate) relay_events: Keyspace, @@ -580,7 +580,7 @@ next_jetstream_id: Arc::new(AtomicU64::new(0)), #[cfg(feature = "jetstream")] last_jetstream_time_us: Arc::new(AtomicI64::new(0)), - #[cfg(feature = "jetstream")] + #[cfg(all(feature = "jetstream", feature = "relay"))] jetstream_lock: Arc::new(parking_lot::Mutex::new(())), #[cfg(feature = "relay")] relay_events, diff --git a/src/ingest/relay.rs b/src/ingest/relay.rs --- a/src/ingest/relay.rs +++ b/src/ingest/relay.rs @@ -34,7 +34,7 @@ #[cfg(feature = "relay")] use crate::types::RelayBroadcast; #[cfg(all(feature = "relay", feature = "jetstream"))] -use crate::types::{JetstreamBroadcast, StoredJetstreamEvent}; +use crate::types::StoredJetstreamEvent; use crate::types::{RepoState, RepoStatus}; use crate::util; use smol_str::{SmolStr, ToSmolStr}; @@ -378,7 +378,7 @@ #[cfg(feature = "jetstream")] let jetstream_did = TrimmedDid::from(&commit.repo).into_static(); - let relay_seq = ctx.queue_emit(|seq| { + let _relay_seq = ctx.queue_emit(|seq| { commit.seq = seq; encode_frame("#commit", &commit) })?; @@ -525,7 +525,7 @@ } #[cfg(feature = "relay")] { - let relay_seq = ctx.queue_emit(|seq| { + let _relay_seq = ctx.queue_emit(|seq| { identity.seq = seq; encode_frame("#identity", &identity) })?; @@ -623,7 +623,7 @@ } #[cfg(feature = "relay")] { - let relay_seq = ctx.queue_emit(|seq| { + let _relay_seq = ctx.queue_emit(|seq| { account.seq = seq; encode_frame("#account", &account) })?; -- tangled.sh