Something went wrong. Try again.
lightweight `com.atproto.sync.listReposByCollection`
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357//! Firehose WebSocket subscriber.//!//! Connects to an ATProto relay's `com.atproto.sync.subscribeRepos` endpoint,//! decodes CBOR-framed events, and dispatches them to per-type handlers.//!//! ## Cursor//!//! Commits are processed asynchronously in a parallel worker pool (see//! [`event_dispatcher`]). The cursor is advanced to the highest seq where all//! preceding seqs are also done ("watermark"), updated once per second. On//! reconnect the cursor is read back so we resume from where we left off.//! Starting without a cursor means starting from live (the relay's current//! head), which is correct for a first-time run.//!//! ## Reconnection//!//! Reconnection is handled here with exponential backoff (1 s … 64 s). A//! graceful server close (`StreamErrorKind::Closed`) skips the backoff and//! reconnects immediately. Application-level decode errors (`Decode`,//! `WrongMessageFormat`) leave the WebSocket connection alive; those are//! logged and counted but do not trigger a reconnect. The backoff counter//! resets to 1 s after any successfully processed event.
mod account_event;pub(crate) mod commit_event;mod event_dispatcher;mod identity_event;mod sync_event;mod validate;
use std::sync::Arc;use std::time::{Duration, Instant};
use futures::StreamExt;use jacquard_api::com_atproto::sync::subscribe_repos::{SubscribeRepos, SubscribeReposMessage};use jacquard_common::url::Host;use jacquard_common::xrpc::SubscriptionExt;use jacquard_common::{StreamErrorKind, TungsteniteClient};use tracing::{debug, info, trace, warn};
use crate::error::Result;use crate::storage::{self, DbRef};use crate::util::TokenExt;use event_dispatcher::CommitDispatcher;
/// Maximum reconnect delay.const MAX_BACKOFF_SECS: u64 = 64;/// Reconnect if no message is received for this long.const IDLE_TIMEOUT: Duration = Duration::from_secs(180);
/// Cap on outstanding firehose events (queued + in-flight) across all DIDs.////// When reached, the WebSocket read is suspended — tungstenite's read buffer/// fills, the TCP receive window closes, and the relay stops sending. Real/// backpressure, no dropped events. The cap resumes as soon as worker/// completions bring outstanding below the threshold.////// Sized to bound peak memory from pending `Box<Commit>` entries (each holding/// a full CAR slice) while leaving plenty of headroom for normal operation.////// Exported as `lightrail_firehose_queue_cap` so Prometheus can compute/// saturation as `lightrail_firehose_outstanding / lightrail_firehose_queue_cap`.const GLOBAL_QUEUE_CAP: usize = 2000;
/// Manages a single logical connection to a relay firehose, with reconnection.pub struct Subscriber { host: Host, db: DbRef, resolver: Arc<crate::identity::Resolver>, max_workers: usize, cursor_save_interval: Duration, client: crate::http::ThrottledClient,}
impl Subscriber { pub fn new( host: Host, db: DbRef, resolver: Arc<crate::identity::Resolver>, max_workers: usize, cursor_save_interval: Duration, client: crate::http::ThrottledClient, ) -> Self { Self { host, db, resolver, max_workers, cursor_save_interval, client, } }
/// Connect and run the subscriber loop, reconnecting on disconnect. /// /// Returns only when the cancellation token fires or an unrecoverable /// storage error occurs. pub async fn run(&mut self, token: tokio_util::sync::CancellationToken) -> Result<()> { metrics::gauge!("lightrail_firehose_queue_cap").set(GLOBAL_QUEUE_CAP as f64); let client = TungsteniteClient::new(); let base: jacquard_common::url::Url = format!("wss://{}", self.host).parse().map_err( |e: jacquard_common::url::ParseError| crate::error::Error::Other(e.to_string()), )?;
let mut backoff_secs: u64 = 1;
// The dispatcher survives reconnects so in-flight workers keep running // and the watermark doesn't regress. let mut dispatcher = CommitDispatcher::new( self.resolver.clone(), self.db.clone(), self.max_workers, self.client.clone(), ); let mut last_seq: i64 = 0; let mut cursor_tick = Instant::now();
'reconnect: loop { if token.is_cancelled() { return Ok(()); }
let connect_cursor: Option<u64> = { let db = self.db.clone(); let host = self.host.clone(); tokio::task::spawn_blocking(move || storage::firehose_cursor::get(&db, &host)) .await?? }; let params = SubscribeRepos { cursor: connect_cursor.map(|c| c as i64), };
info!( host = %self.host, cursor = ?connect_cursor, "connecting to firehose", );
let sub = client.subscription(base.clone()); let Some(stream) = token.run(sub.subscribe(¶ms)).await else { return Ok(()); };
let stream = match stream { Ok(s) => s, Err(e) => { warn!( error = %e, host = %self.host, backoff_secs, "firehose connect failed; will retry", ); if !token.sleep(Duration::from_secs(backoff_secs)).await { return Ok(()); } backoff_secs = (backoff_secs * 2).min(MAX_BACKOFF_SECS); continue 'reconnect; } };
info!(host = %self.host, "firehose connected");
// `_sink` kept alive so the write half of the WebSocket isn't dropped. let (_sink, mut messages) = stream.into_stream(); let mut last_message_at = Instant::now(); let mut first_seq_logged = false;
loop { // Unified select! driving both the WebSocket stream and worker // completions. We use an enum to keep the match arms clean. enum Event { Stream( Option< std::result::Result< SubscribeReposMessage<'static>, jacquard_common::StreamError, >, >, ), Worker( Option< std::result::Result< (tokio::task::Id, event_dispatcher::CommitWorkerResult), tokio::task::JoinError, >, >, ), Timeout, }
// Gate the stream branch on queue capacity (see GLOBAL_QUEUE_CAP). let can_accept = dispatcher.outstanding_count() < GLOBAL_QUEUE_CAP; if !can_accept { metrics::counter!("lightrail_firehose_backpressure_ticks_total").increment(1); }
let event = tokio::select! { _ = token.cancelled() => return Ok(()), r = messages.next(), if can_accept => Event::Stream(r), r = dispatcher.workers_mut().join_next_with_id(), if dispatcher.has_workers() => Event::Worker(r), _ = tokio::time::sleep( IDLE_TIMEOUT.saturating_sub(last_message_at.elapsed()) ) => Event::Timeout, };
match event { Event::Stream(None) => { info!(host = %self.host, "firehose stream ended; reconnecting"); continue 'reconnect; } Event::Stream(Some(Err(e))) => { last_message_at = Instant::now(); match e.kind() { // Application-level decode failure on a healthy // WebSocket connection: jacquard's filter_map stream // continues yielding after this error, so just log, // count, and continue reading. StreamErrorKind::Decode | StreamErrorKind::WrongMessageFormat => { metrics::counter!("lightrail_firehose_decode_errors_total") .increment(1); warn!( error = %e, host = %self.host, "firehose message decode error; skipping frame", ); // No reconnect — the WebSocket connection is // still alive. } // Graceful server close: reconnect immediately. StreamErrorKind::Closed => { info!(host = %self.host, "firehose stream closed; reconnecting"); continue 'reconnect; } // Transport / protocol error: the connection is // broken; reconnect with backoff. _ => { warn!( error = %e, host = %self.host, backoff_secs, "firehose stream error; will reconnect", ); if !token.sleep(Duration::from_secs(backoff_secs)).await { return Ok(()); }; backoff_secs = (backoff_secs * 2).min(MAX_BACKOFF_SECS); continue 'reconnect; } } } Event::Stream(Some(Ok(msg))) => { last_message_at = Instant::now(); backoff_secs = 1; if let Some(seq) = self.dispatch(msg, &mut dispatcher) { if !first_seq_logged { first_seq_logged = true; let gap = connect_cursor .map(|c| format!("{}", seq - c as i64)) .unwrap_or("[no connect cursor]".to_string()); info!( host = %self.host, cursor = ?connect_cursor, first_seq = seq, gap, "firehose replay started", ); } last_seq = seq; } } Event::Worker(Some(Ok((id, result)))) => { dispatcher.handle_completion(id, Ok(result)); } Event::Worker(Some(Err(join_err))) => { dispatcher.handle_completion(join_err.id(), Err(join_err)); } Event::Worker(None) => {} // JoinSet drained (shouldn't happen with guard) Event::Timeout => { warn!( host = %self.host, timeout_secs = IDLE_TIMEOUT.as_secs(), "firehose idle timeout; reconnecting", ); continue 'reconnect; } }
if cursor_tick.elapsed() >= self.cursor_save_interval { dispatcher.evict_stalled(); let wm = dispatcher.watermark(last_seq); let db = self.db.clone(); let host = self.host.clone(); tokio::task::spawn_blocking(move || { storage::firehose_cursor::set(&db, &host, wm as u64) }) .await??; cursor_tick = Instant::now(); } } } }
// ----------------------------------------------------------------------- // Dispatch // -----------------------------------------------------------------------
/// Route a decoded message to the appropriate handler. /// /// Returns the sequence number to use as the stream position, or `None` /// for `#info` messages (which carry no seq). fn dispatch( &mut self, msg: SubscribeReposMessage<'static>, dispatcher: &mut CommitDispatcher, ) -> Option<i64> { let seq = match &msg { SubscribeReposMessage::Commit(c) => c.seq, SubscribeReposMessage::Sync(s) => s.seq, SubscribeReposMessage::Identity(i) => i.seq, SubscribeReposMessage::Account(a) => a.seq, // #info and unknown frames (e.g. server error frames like FutureCursor) // carry no sequence number and don't advance the cursor. SubscribeReposMessage::Info(info) => { info!(name = %info.name, message = ?info.message, "firehose #info"); return None; } SubscribeReposMessage::Unknown(_) => { info!("firehose unknown frame"); return None; } };
match msg { SubscribeReposMessage::Commit(commit) => { trace!(did = %commit.repo, rev = %commit.rev, seq, "firehose #commit"); dispatcher.enqueue(commit, seq); } SubscribeReposMessage::Sync(sync) => { debug!(did = %sync.did, seq, "firehose #sync"); dispatcher.enqueue_sync(sync, seq); } SubscribeReposMessage::Account(account) => { debug!(did = %account.did, active = account.active, seq, "firehose #account"); dispatcher.enqueue_account(account, seq); } SubscribeReposMessage::Identity(identity) => { debug!(did = %identity.did, seq, "firehose #identity"); dispatcher.enqueue_identity(identity, seq); } SubscribeReposMessage::Info(_) | SubscribeReposMessage::Unknown(_) => {} }
Some(seq) }}