[crawler] rewrite component, spread into producers, add consuming from listReposByCollection

README.md

···
 
 `hydrant` is an AT Protocol indexer built on the `fjall` database that handles sync for you. it's flexible, supporting both full-network indexing and filtered indexing (e.g., by DID), also allowing querying with XRPCs and providing an ordered event stream with cursor support.
 
-you can see [random.wisp.place](https://tangled.org/did:plc:dfl62fgb7wtjj3fcbb72naae/random.wisp.place) for an example on how to use hydrant.
+you can see [random.wisp.place](https://tangled.org/did:plc:dfl62fgb7wtjj3fcbb72naae/random.wisp.place) (standalone binary using http API) or the [statusphere example](./examples/statusphere.rs) (hydrant-as-library) for examples on how to use hydrant.
 
 **WARNING: *the db format is not stable yet.*** it's in active development so if you are going to rely on the db format being stable, don't (eg. for query features, if you are using ephemeral mode this doesn't matter for example, or you dont mind losing your existing backfilled data in hydrant if you already processed them.).
 
 ## vs `tap`
 
-while [`tap`](https://github.com/bluesky-social/indigo/tree/main/cmd/tap) is designed as a firehose consumer and simply just propagates events while handling sync, `hydrant` is flexible, it allows you to directly query the database for records, and it also provides an ordered view of events, allowing the use of a cursor to fetch events from a specific point in time.
+while [`tap`](https://github.com/bluesky-social/indigo/tree/main/cmd/tap) is designed as a firehose consumer and simply just propagates events while handling sync, `hydrant` is flexible, it allows you to directly query the database for records, and it also provides an ordered view of events, allowing the use of a cursor to fetch events from a specific point. it can act as both an indexer or an ephemeral view of some window of events.
 
 ### stream behavior
 
···
 
 ### multiple relay support
 
-`hydrant` supports connecting to multiple relays simultaneously for both firehose ingestion and crawling. when `RELAY_HOSTS` is configured with multiple URLs:
+`hydrant` supports connecting to multiple relays simultaneously for firehose ingestion. when `RELAY_HOSTS` is configured with multiple URLs:
 
 - one independent firehose stream loop is spawned per relay
-- one independent crawling loop is spawned per relay
-- each relay maintains its own firehose / crawler cursor state
-- all ingestion loops and crawlers share the same worker pool and database
-- all crawlers share the same pending queue for backfill
+- each relay maintains its own firehose cursor state
+- all ingestion loops share the same worker pool and database
 
 commit events are de-duplicated according to the repo `rev`. account / identity events are de-duplicated using the `time` field.
 todo: decide what to do on relay-side account takedowns or if relays set the `time` field.
 
+### crawler sources
+
+the crawler is configured separately from the firehose via `CRAWLER_URLS`. each source is a `[mode::]url` entry where the mode prefix is optional and defaults to `by_collection` in filter mode or `relay` in full-network mode.
+
+- `relay`: enumerates the network via `com.atproto.sync.listRepos`, then checks each repo's collections via `describeRepo`. used for full-network discovery.
+- `by_collection`: queries `com.atproto.sync.listReposByCollection` for each configured signal. more efficient for filtered indexing since it only surfaces repos that have matching records.
+cursors are stored per collection.
+
+```
+CRAWLER_URLS=by_collection::https://lightrail.microcosm.blue,relay::wss://bsky.network
+```
+
+each source maintains its own cursor so restarts resume mid-pass.
+
 ## configuration
 
 `hydrant` is configured via environment variables. all variables are prefixed with `HYDRANT_` (except `RUST_LOG`).
···
 | :--- | :--- | :--- |
 | `DATABASE_PATH` | `./hydrant.db` | path to the database folder. |
 | `RUST_LOG` | `info` | log filter directives (e.g., `debug`, `hydrant=trace`). [`tracing` env-filter syntax](https://docs.rs/tracing-subscriber/latest/tracing_subscriber/filter/struct.EnvFilter.html). |
-| `RELAY_HOST` | `wss://relay.fire.hose.cam/` | URL of the relay. |
-| `RELAY_HOSTS` | | comma-separated list of relay URLs. if unset, falls back to `RELAY_HOST`. |
+| `RELAY_HOST` | `wss://relay.fire.hose.cam/` | URL of the relay (firehose only). |
+| `RELAY_HOSTS` | | comma-separated list of relay URLs (firehose only). if unset, falls back to `RELAY_HOST`. |
+| `CRAWLER_URLS` | relay hosts in full-network mode, `https://lightrail.microcosm.blue` in filter mode | comma-separated list of `[mode::]url` crawler sources. mode is `relay` or `by_collection`; bare URLs use the default mode. set to empty string to disable crawling. |
 | `PLC_URL` | `https://plc.wtf`, `https://plc.directory` if full network | base URL(s) of the PLC directory (comma-separated for multiple). |
 | `EPHEMERAL` | `false` | if enabled, no records are stored. events are deleted after a certain duration (`EPHEMERAL_TTL`). |
 | `EPHEMERAL_TTL` | `60min` | decides after how long events should be deleted. |
···
 | `ENABLE_DEBUG` | `false` | enable debug endpoints. |
 | `DEBUG_PORT` | `API_PORT + 1` | port for debug endpoints (if enabled). |
 | `ENABLE_FIREHOSE` | `true` | whether to ingest relay subscriptions. |
-| `ENABLE_CRAWLER` | `false` (if Filter), `true` (if Full) | whether to actively query the network for unknown repositories. |
+| `ENABLE_CRAWLER` | `true` if full network or crawler sources are configured, `false` otherwise | whether to actively query the network for unknown repositories. |
 | `CRAWLER_MAX_PENDING_REPOS` | `2000` | max pending repos for crawler. |
 | `CRAWLER_RESUME_PENDING_REPOS` | `1000` | resume threshold for crawler pending repos. |
 
···
 
 - `POST /db/train`: train zstd compression dictionaries for the `repos`, `blocks`, and `events` keyspaces. dictionaries are written to disk; a restart is required to apply them. the crawler, firehose, and backfill worker are paused for the duration and restored on completion.
 - `POST /db/compact`: trigger a full major compaction of all database keyspaces in parallel. the crawler, firehose, and backfill worker are paused for the duration and restored on completion.
+- `DELETE /cursors`: reset all stored cursors for a given URL. body: `{ "key": "..." }` where key is a URL. clears the relay crawler cursor, and any by-collection cursors associated with that URL. causes the next crawler pass to restart from the beginning.
 
 #### filter mode
 

examples/statusphere.rs

···
 //!
 //! the database persists records across restarts. on each start the full event
 //! history is replayed from the database to rebuild the in-memory index.
+//! (in a better app, we could for example use the ephemeral mode of hydrant,
+//! and use our db, or we could use hydrant to backfill multiple instances of the app.)
 
+use std::str::FromStr;
 use std::sync::Arc;
 use std::time::Duration;
 
+use chrono::DateTime;
 use futures::StreamExt;
 use hydrant::config::Config;
 use hydrant::control::{EventStream, Hydrant, ReposControl};
 use hydrant::filter::FilterMode;
+use jacquard_common::types::tid::Tid;
 use scc::HashMap;
 
 const COLLECTION: &str = "xyz.statusphere.status";
···
         }
     }
 
-    fn set(&self, did: String, emoji: String, created_at: String) -> bool {
+    fn set(&self, did: String, emoji: String, created_at: &str) -> bool {
         let is_newer = self
             .current
-            .read_sync(&did, |_, e| created_at > e.created_at)
+            .read_sync(&did, |_, e| created_at > e.created_at.as_str())
             .unwrap_or(true);
         if is_newer {
-            self.current
-                .upsert_sync(did, StatusEntry { emoji, created_at });
+            self.current.upsert_sync(
+                did,
+                StatusEntry {
+                    emoji,
+                    created_at: created_at.to_owned(),
+                },
+            );
         }
         is_newer
     }
···
                     let created_at = record
                         .get("createdAt")
                         .and_then(|v| v.as_str())
-                        .unwrap_or("")
-                        .to_owned();
+                        .unwrap_or("");
                     if index.set(did.clone(), emoji.clone(), created_at) {
                         let name = repos
                             .get(&rec.did)
···
                             .flatten()
                             .and_then(|info| info.handle)
                             .unwrap_or(did);
-                        println!("[{}] {name} set status: {emoji}", event.id);
+                        println!("[{created_at}] {name}: {emoji}");
                     }
                 }
                 "delete" => {
···
                         .and_then(|info| info.handle)
                         .unwrap_or(did.clone());
                     index.delete(&did);
-                    println!("[{}] {name} cleared status", event.id);
+                    let date = Tid::from_str(&rec.rkey)
+                        .ok()
+                        .and_then(|tid| DateTime::from_timestamp_micros(tid.timestamp() as i64))
+                        .map(|date| date.to_string())
+                        .unwrap_or_else(|| "invalid rkey".to_string());
+                    println!("[{date}] {name} cleared status");
                 }
                 _ => {}
             }
···
         .with_env_filter("hydrant=info")
         .init();
 
+    // config is loaded from environment variables (all prefixed with HYDRANT_).
+    // key defaults for this example:
+    // DATABASE_PATH=./hydrant.db                    | where to store the database.
+    // RELAY_HOST=wss://relay.fire.hose.cam/         | firehose source.
+    // CRAWLER_URLS=https://lightrail.microcosm.blue | crawler sources. in filter mode this defaults to `by-collection`.
     let cfg = Config::from_env()?;
     let hydrant = Hydrant::new(cfg).await?;
 

src/api/db.rs

···
 use crate::control::Hydrant;
-use axum::{Router, extract::State, http::StatusCode, routing::post};
+use axum::{
+    Json, Router,
+    extract::State,
+    http::StatusCode,
+    routing::{delete, post},
+};
+use serde::Deserialize;
 
 pub fn router() -> Router<Hydrant> {
     Router::new()
         .route("/db/train", post(handle_train_dict))
         .route("/db/compact", post(handle_compact))
+        .route("/cursors", delete(handle_reset_cursor))
 }
 
 pub async fn handle_train_dict(
···
     hydrant
         .db
         .train_dicts()
+        .await
+        .map_err(|e| (StatusCode::INTERNAL_SERVER_ERROR, e.to_string()))?;
+    Ok(StatusCode::OK)
+}
+
+#[derive(Deserialize)]
+pub struct ResetCursorBody {
+    pub key: String,
+}
+
+pub async fn handle_reset_cursor(
+    State(hydrant): State<Hydrant>,
+    Json(body): Json<ResetCursorBody>,
+) -> Result<StatusCode, (StatusCode, String)> {
+    hydrant
+        .crawler
+        .reset_cursor(&body.key)
         .await
         .map_err(|e| (StatusCode::INTERNAL_SERVER_ERROR, e.to_string()))?;
     Ok(StatusCode::OK)

src/config.rs

···
 use url::Url;
 
 #[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub enum CrawlerMode {
+    /// enumerate via `com.atproto.sync.listRepos`, then check signals with `describeRepo`.
+    Relay,
+    /// enumerate via `com.atproto.sync.listReposByCollection` for each configured signal.
+    ByCollection,
+}
+
+impl CrawlerMode {
+    fn default_for(full_network: bool) -> Self {
+        if full_network {
+            Self::Relay
+        } else {
+            Self::ByCollection
+        }
+    }
+}
+
+impl FromStr for CrawlerMode {
+    type Err = miette::Error;
+    fn from_str(s: &str) -> Result<Self> {
+        match s {
+            "relay" => Ok(Self::Relay),
+            "by_collection" | "by-collection" => Ok(Self::ByCollection),
+            _ => Err(miette::miette!(
+                "invalid crawler mode: expected 'relay' or 'by_collection'"
+            )),
+        }
+    }
+}
+
+impl fmt::Display for CrawlerMode {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        match self {
+            Self::Relay => write!(f, "relay"),
+            Self::ByCollection => write!(f, "by_collection"),
+        }
+    }
+}
+
+/// a single crawler source: a URL and the mode used to enumerate it.
+#[derive(Debug, Clone)]
+pub struct CrawlerSource {
+    pub url: Url,
+    pub mode: CrawlerMode,
+}
+
+impl CrawlerSource {
+    /// parse `[mode::]url` — mode prefix is optional, falls back to `default_mode`.
+    fn parse(s: &str, default_mode: CrawlerMode) -> Option<Self> {
+        if let Some((prefix, rest)) = s.split_once("::") {
+            let mode = prefix.parse().ok()?;
+            let url = Url::parse(rest).ok()?;
+            Some(Self { url, mode })
+        } else {
+            let url = Url::parse(s).ok()?;
+            Some(Self {
+                url,
+                mode: default_mode,
+            })
+        }
+    }
+}
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
 pub enum Compression {
     Lz4,
     Zstd,
···
     pub filter_signals: Option<Vec<String>>,
     pub filter_collections: Option<Vec<String>>,
     pub filter_excludes: Option<Vec<String>>,
+    /// crawler sources: each entry pairs a URL with a discovery mode.
+    ///
+    /// set via `HYDRANT_CRAWLER_URLS` as a comma-separated list of `[mode::]url` entries,
+    /// e.g. `relay::wss://bsky.network,by_collection::https://lightrail.microcosm.blue`.
+    /// a bare URL without a `mode::` prefix uses the default mode (`relay` for full-network,
+    /// `by_collection` otherwise). defaults to the relay hosts with the default mode.
+    /// set to an empty string to disable crawling entirely.
+    pub crawler_sources: Vec<CrawlerSource>,
 }
 
 impl Config {
···
                 .collect()
         });
 
+        let default_mode = CrawlerMode::default_for(full_network);
+        let crawler_sources = match std::env::var("HYDRANT_CRAWLER_URLS") {
+            Ok(s) => s
+                .split(',')
+                .map(|s| s.trim())
+                .filter(|s| !s.is_empty())
+                .filter_map(|s| CrawlerSource::parse(s, default_mode))
+                .collect(),
+            Err(_) => match default_mode {
+                CrawlerMode::Relay => relay_hosts
+                    .iter()
+                    .map(|url| CrawlerSource {
+                        url: url.clone(),
+                        mode: CrawlerMode::Relay,
+                    })
+                    .collect(),
+                CrawlerMode::ByCollection => vec![CrawlerSource {
+                    url: Url::parse("https://lightrail.microcosm.blue").unwrap(),
+                    mode: CrawlerMode::ByCollection,
+                }],
+            },
+        };
+
         Ok(Self {
             database_path,
             relays: relay_hosts,
···
             filter_signals,
             filter_collections,
             filter_excludes,
+            crawler_sources,
         })
     }
 }
···
             "crawler resume pending",
             self.crawler_resume_pending_repos
         )?;
+        if !self.crawler_sources.is_empty() {
+            let sources: Vec<_> = self
+                .crawler_sources
+                .iter()
+                .map(|s| format!("{}::{}", s.mode, s.url))
+                .collect();
+            config_line!(f, "crawler sources", sources.join(", "))?;
+        }
         if let Some(signals) = &self.filter_signals {
             config_line!(f, "filter signals", format_args!("{:?}", signals))?;
         }

src/control.rs

···
 
 use crate::backfill::BackfillWorker;
 use crate::config::{Config, SignatureVerification};
-use crate::crawler::Crawler;
 use crate::db::{self, filter as db_filter, keys, ser_repo_state};
 use crate::filter::{FilterMode, SetUpdate};
 use crate::ingest::{firehose::FirehoseIngestor, worker::FirehoseWorker};
···
         // 4. set crawler enabled state from config, evaluated against the post-patch filter
         let post_patch_crawler = match config.enable_crawler {
             Some(b) => b,
-            None => state.filter.load().mode == FilterMode::Full,
+            None => {
+                state.filter.load().mode == FilterMode::Full || !config.crawler_sources.is_empty()
+            }
         };
         state.crawler_enabled.send_replace(post_patch_crawler);
 
···
                 }
             }
 
-            // 11. spawn the crawler if we have relay hosts to crawl
-            if !relay_hosts.is_empty() {
-                let crawler_rx = state.crawler_enabled.subscribe();
+            // 11. spawn crawler components
+            if !config.crawler_sources.is_empty() {
+                use crate::config::CrawlerMode;
+                use crate::crawler::throttle::Throttler;
+                use crate::crawler::{
+                    ByCollectionProducer, CrawlerStats, CrawlerWorker, InFlight, RelayProducer,
+                    RetryProducer, SignalChecker,
+                };
+                use std::time::Duration;
+                use tracing::Instrument;
+
+                let http = reqwest::Client::builder()
+                    .user_agent(concat!(
+                        env!("CARGO_PKG_NAME"),
+                        "/",
+                        env!("CARGO_PKG_VERSION")
+                    ))
+                    .gzip(true)
+                    .build()
+                    .expect("that reqwest will build");
+                let pds_throttler = Throttler::new();
+                let in_flight = InFlight::new();
+                let stats = CrawlerStats::new(
+                    state.clone(),
+                    config
+                        .crawler_sources
+                        .iter()
+                        .map(|s| s.url.clone())
+                        .collect(),
+                    pds_throttler.clone(),
+                );
+                let checker = SignalChecker {
+                    http: http.clone(),
+                    state: state.clone(),
+                    throttler: pds_throttler,
+                };
+
                 info!(
-                    relay_count = relay_hosts.len(),
-                    hosts = relay_hosts
-                        .iter()
-                        .map(|h| h.as_str())
-                        .collect::<Vec<_>>()
-                        .join(", "),
+                    max_pending = config.crawler_max_pending_repos,
+                    resume_pending = config.crawler_resume_pending_repos,
                     enabled = *state.crawler_enabled.borrow(),
-                    "starting crawler(s)"
+                    "starting crawler worker"
                 );
-                let state = state.clone();
-                let max_pending = config.crawler_max_pending_repos;
-                let resume_pending = config.crawler_resume_pending_repos;
+                let (worker, tx) = CrawlerWorker::new(
+                    state.clone(),
+                    config.crawler_max_pending_repos,
+                    config.crawler_resume_pending_repos,
+                    stats.clone(),
+                );
+                tokio::spawn(async move {
+                    worker.run().await;
+                    error!("crawler worker exited unexpectedly, aborting");
+                    std::process::abort();
+                });
+
+                let ticker = tokio::spawn(stats.clone().task());
                 tokio::spawn(async move {
-                    let crawler =
-                        Crawler::new(state, relay_hosts, max_pending, resume_pending, crawler_rx);
-                    if let Err(e) = crawler.run().await {
-                        error!(err = %e, "crawler error");
-                        db::check_poisoned_report(&e);
+                    match ticker.await {
+                        Err(e) => error!(err = ?e, "stats ticker panicked, aborting"),
+                        Ok(()) => error!("stats ticker exited unexpectedly, aborting"),
                     }
+                    std::process::abort();
                 });
+
+                tokio::spawn(
+                    RetryProducer {
+                        checker: checker.clone(),
+                        in_flight: in_flight.clone(),
+                        tx: tx.clone(),
+                    }
+                    .run(),
+                );
+
+                let crawler_rx = state.crawler_enabled.subscribe();
+                for source in config.crawler_sources.iter().cloned() {
+                    let http = http.clone();
+                    let state = state.clone();
+                    let in_flight = in_flight.clone();
+                    let tx = tx.clone();
+                    let stats = stats.clone();
+                    let enabled = crawler_rx.clone();
+                    match source.mode {
+                        CrawlerMode::Relay => {
+                            info!(relay = %source.url, enabled = *state.crawler_enabled.borrow(), "starting relay crawler");
+                            let span = tracing::info_span!("crawl", url = %source.url);
+                            tokio::spawn(
+                                RelayProducer {
+                                    relay_url: source.url,
+                                    checker: checker.clone(),
+                                    in_flight,
+                                    tx,
+                                    enabled,
+                                    stats,
+                                }
+                                .run()
+                                .instrument(span),
+                            );
+                        }
+                        CrawlerMode::ByCollection => {
+                            info!(
+                                host = source.url.host_str(),
+                                enabled = *state.crawler_enabled.borrow(),
+                                "starting by-collection crawler"
+                            );
+                            let span =
+                                tracing::info_span!("by_collection", host = source.url.host_str());
+                            tokio::spawn(
+                                async move {
+                                    loop {
+                                        let producer = ByCollectionProducer {
+                                            index_url: source.url.clone(),
+                                            http: http.clone(),
+                                            state: state.clone(),
+                                            in_flight: in_flight.clone(),
+                                            tx: tx.clone(),
+                                            enabled: enabled.clone(),
+                                            stats: stats.clone(),
+                                        };
+                                        if let Err(e) = producer.run().await {
+                                            error!(err = ?e, "by-collection crawler fatal error, restarting in 30s");
+                                            tokio::time::sleep(Duration::from_secs(30)).await;
+                                        }
+                                    }
+                                }
+                                .instrument(span),
+                            );
+                        }
+                    }
+                }
             }
 
             // 12. spawn the firehose worker on a blocking thread (fatal task)
···
 pub struct CrawlerHandle(Arc<AppState>);
 
 impl CrawlerHandle {
-    /// enable the crawler. no-op if already enabled.
+    /// enable the crawler (enables all configured producers). no-op if already enabled.
     pub fn enable(&self) {
         self.0.crawler_enabled.send_replace(true);
     }
-    /// disable the crawler. in-progress repo checks finish before the crawler pauses.
+    /// disable the crawler (disables all configured producers).
+    /// in-progress repo checks finish before the crawler pauses.
     pub fn disable(&self) {
         self.0.crawler_enabled.send_replace(false);
     }
···
     pub fn is_enabled(&self) -> bool {
         *self.0.crawler_enabled.borrow()
     }
+
+    /// delete all cursor entries associated with the given URL.
+    pub async fn reset_cursor(&self, url: &str) -> Result<()> {
+        let db = self.0.db.clone();
+        let point_keys = [keys::crawler_cursor_key(url)];
+        let by_collection_prefix = keys::by_collection_cursor_prefix(url);
+        tokio::task::spawn_blocking(move || {
+            let mut batch = db.inner.batch();
+            for k in point_keys {
+                batch.remove(&db.cursors, k);
+            }
+            for entry in db.cursors.prefix(&by_collection_prefix) {
+                let (k, _) = entry.into_inner().into_diagnostic()?;
+                batch.remove(&db.cursors, k);
+            }
+            batch.commit().into_diagnostic()
+        })
+        .await
+        .into_diagnostic()??;
+        Ok(())
+    }
 }
 
 /// runtime control over the firehose ingestor component.
-///
-/// the firehose connects to each configured relay's `com.atproto.sync.subscribeRepos`
-/// websocket and processes commit, identity, account, and sync events in real time.
-/// one independent connection is maintained per relay URL.
-///
-/// disabling the firehose closes the websocket after the current message is processed.
 #[derive(Clone)]
 pub struct FirehoseHandle(Arc<AppState>);
 
···
 /// the backfill worker fetches full repo CAR files from each repo's PDS for any
 /// repository in the pending queue, parses the MST, and inserts all matching records
 /// into the database. concurrency is bounded by `HYDRANT_BACKFILL_CONCURRENCY_LIMIT`.
-///
-/// disabling backfill lets any in-flight repo fetches finish before pausing.
 #[derive(Clone)]
 pub struct BackfillHandle(Arc<AppState>);
 
···
     pub fn enable(&self) {
         self.0.backfill_enabled.send_replace(true);
     }
-    /// disable the backfill worker. in-flight repo fetches complete before pausing.
+    /// disable the backfill worker. in-flight repos complete before pausing.
     pub fn disable(&self) {
         self.0.backfill_enabled.send_replace(false);
     }
···
         *self.0.backfill_enabled.borrow()
     }
 }
-
-// --- filter control ---
 
 /// a point-in-time snapshot of the filter configuration. returned by all [`FilterControl`] methods.
 ///

src/crawler/by_collection.rs

···
+use crate::db::keys::by_collection_cursor_key;
+use crate::db::{Db, keys};
+use crate::state::AppState;
+use crate::util::{ErrorForStatus, RetryOutcome, RetryWithBackoff, WatchEnabledExt};
+use jacquard_api::com_atproto::sync::list_repos_by_collection::ListReposByCollectionOutput;
+use jacquard_common::{IntoStatic, types::string::Did};
+use miette::{Context, IntoDiagnostic, Result};
+use smol_str::SmolStr;
+use std::sync::Arc;
+use std::time::Duration;
+use tokio::sync::{mpsc, watch};
+use tracing::{debug, error, info, trace, warn};
+use url::Url;
+
+use super::worker::{CrawlerBatch, CursorUpdate};
+use super::{CrawlerStats, InFlight, base_url};
+
+const BLOCKING_TASK_TIMEOUT: Duration = Duration::from_secs(30);
+
+pub(crate) struct ByCollectionProducer {
+    pub(crate) index_url: Url,
+    pub(crate) http: reqwest::Client,
+    pub(crate) state: Arc<AppState>,
+    pub(crate) in_flight: InFlight,
+    pub(crate) tx: mpsc::Sender<CrawlerBatch>,
+    pub(crate) enabled: watch::Receiver<bool>,
+    pub(crate) stats: CrawlerStats,
+}
+
+impl ByCollectionProducer {
+    /// hourly loop: runs one full pass per configured signal, then sleeps.
+    pub(crate) async fn run(mut self) -> Result<()> {
+        loop {
+            self.enabled.wait_enabled("by-collection crawler").await;
+
+            let filter = self.state.filter.load();
+            if filter.signals.is_empty() {
+                debug!("no signals configured, by-collection crawler sleeping 5m");
+                tokio::time::sleep(Duration::from_secs(300)).await;
+                continue;
+            }
+
+            let signals: Vec<String> = filter.signals.iter().map(|s| s.to_string()).collect();
+            drop(filter);
+
+            info!(
+                host = self.index_url.host_str(),
+                signal_count = signals.len(),
+                "starting by-collection discovery pass"
+            );
+
+            for collection in &signals {
+                self.enabled.wait_enabled("by-collection crawler").await;
+                let span = tracing::info_span!("by_collection", %collection);
+                use tracing::Instrument as _;
+                if let Err(e) = self.crawl(collection).instrument(span).await {
+                    error!(err = %e, %collection, "by-collection crawl error, continuing");
+                }
+            }
+
+            info!("by-collection pass complete, sleeping 1h");
+            tokio::select! {
+                _ = tokio::time::sleep(Duration::from_secs(3600)) => {}
+                _ = self.enabled.changed() => {
+                    if !*self.enabled.borrow() { return Ok(()); }
+                }
+            }
+        }
+    }
+
+    async fn crawl(&mut self, collection: &str) -> Result<()> {
+        let db = &self.state.db;
+        let base = base_url(&self.index_url)?;
+        let cursor_key = by_collection_cursor_key(self.index_url.as_str(), collection);
+
+        // resume from any persisted cursor, so a restart mid-pass doesn't rescan from scratch.
+        let mut cursor: Option<SmolStr> = Db::get(db.cursors.clone(), &cursor_key)
+            .await
+            .ok()
+            .flatten()
+            .and_then(|b| String::from_utf8(b.to_vec()).ok().map(SmolStr::from));
+
+        if cursor.is_some() {
+            debug!(%collection, "resuming by-collection pass from cursor");
+        }
+
+        let mut total_count = 0usize;
+        let mut new_count = 0usize;
+
+        loop {
+            self.enabled.wait_enabled("by-collection crawler").await;
+
+            let mut url = base
+                .join("/xrpc/com.atproto.sync.listReposByCollection")
+                .into_diagnostic()?;
+            url.query_pairs_mut()
+                .append_pair("collection", collection)
+                .append_pair("limit", "2000");
+            if let Some(c) = &cursor {
+                url.query_pairs_mut().append_pair("cursor", c.as_str());
+            }
+
+            let fetch_result = (|| self.http.get(url.clone()).send().error_for_status())
+                .retry(5, |e: &reqwest::Error, attempt| {
+                    matches!(e.status(), Some(reqwest::StatusCode::TOO_MANY_REQUESTS))
+                        .then(|| Duration::from_secs(1 << attempt.min(5)))
+                })
+                .await;
+
+            let res = match fetch_result {
+                Ok(r) => r,
+                Err(RetryOutcome::Ratelimited) => {
+                    warn!(%collection, "rate limited by collection index after retries");
+                    continue;
+                }
+                Err(RetryOutcome::Failed(e)) => {
+                    error!(err = %e, %collection, "by-collection fetch failed");
+                    continue;
+                }
+            };
+
+            let bytes = match res.bytes().await {
+                Ok(b) => b,
+                Err(e) => {
+                    error!(err = %e, "can't read listReposByCollection response");
+                    continue;
+                }
+            };
+
+            struct PageResult {
+                unknown_dids: Vec<Did<'static>>,
+                next_cursor: Option<SmolStr>,
+                count: usize,
+            }
+
+            let page_result = {
+                let repos_ks = db.repos.clone();
+                let filter_ks = db.filter.clone();
+
+                tokio::time::timeout(
+                    BLOCKING_TASK_TIMEOUT,
+                    tokio::task::spawn_blocking(move || -> miette::Result<Option<PageResult>> {
+                        let output =
+                            match serde_json::from_slice::<ListReposByCollectionOutput>(&bytes) {
+                                Ok(out) => out.into_static(),
+                                Err(e) => {
+                                    error!(
+                                        err = %e,
+                                        "failed to parse listReposByCollection response"
+                                    );
+                                    return Ok(None);
+                                }
+                            };
+
+                        if output.repos.is_empty() {
+                            return Ok(None);
+                        }
+
+                        let count = output.repos.len();
+                        let next_cursor = output.cursor.map(|c| c.as_str().into());
+                        let mut unknown = Vec::new();
+                        for repo in output.repos {
+                            let excl_key = crate::db::filter::exclude_key(repo.did.as_str())?;
+                            if filter_ks.contains_key(&excl_key).into_diagnostic()? {
+                                continue;
+                            }
+                            let did_key = keys::repo_key(&repo.did);
+                            if !repos_ks.contains_key(&did_key).into_diagnostic()? {
+                                unknown.push(repo.did.into_static());
+                            }
+                        }
+
+                        Ok(Some(PageResult {
+                            unknown_dids: unknown,
+                            next_cursor,
+                            count,
+                        }))
+                    }),
+                )
+                .await
+            }
+            .into_diagnostic()?
+            .map_err(|_| {
+                error!(
+                    "spawn_blocking task for listReposByCollection timed out after {}s",
+                    BLOCKING_TASK_TIMEOUT.as_secs()
+                );
+                miette::miette!("spawn_blocking task for listReposByCollection timed out")
+            })?;
+
+            let PageResult {
+                unknown_dids,
+                next_cursor,
+                count,
+            } = match page_result {
+                Ok(Some(r)) => r,
+                Ok(None) => {
+                    info!(
+                        %collection,
+                        total = total_count,
+                        new = new_count,
+                        "finished by-collection pass (empty page)"
+                    );
+                    return Ok(());
+                }
+                Err(e) => return Err(e).wrap_err("error parsing by-collection page"),
+            };
+
+            total_count += count;
+            trace!(%collection, count, "fetched repos from by-collection index");
+
+            let in_flight = self.in_flight.acquire(unknown_dids).await;
+            new_count += in_flight.len();
+            self.stats.record_crawled(count);
+
+            cursor = next_cursor;
+            let cursor_update = cursor.as_ref().map(|c| CursorUpdate {
+                key: cursor_key.clone(),
+                value: c.as_bytes().to_vec(),
+            });
+
+            self.tx
+                .send(CrawlerBatch {
+                    guards: in_flight,
+                    cursor_update,
+                })
+                .await
+                .ok();
+
+            if cursor.is_none() {
+                info!(
+                    %collection,
+                    total = total_count,
+                    new = new_count,
+                    "finished by-collection pass"
+                );
+                return Ok(());
+            }
+        }
+    }
+}

src/crawler/mod.rs

···
-use crate::crawler::throttle::ThrottleHandle;
-use crate::db::keys::crawler_cursor_key;
-use crate::db::{Db, keys, ser_repo_state};
 use crate::state::AppState;
-use crate::types::RepoState;
-use crate::util::WatchEnabledExt;
-use crate::util::{ErrorForStatus, RetryOutcome, RetryWithBackoff, parse_retry_after};
-use chrono::{DateTime, TimeDelta, Utc};
-use futures::FutureExt;
-use jacquard_api::com_atproto::repo::describe_repo::DescribeRepoOutput;
-use jacquard_api::com_atproto::sync::list_repos::ListReposOutput;
-use jacquard_common::{IntoStatic, types::string::Did};
-use miette::{Context, IntoDiagnostic, Result};
-use rand::Rng;
-use rand::RngExt;
-use rand::rngs::SmallRng;
-use reqwest::StatusCode;
+use futures::future::join_all;
+use jacquard_common::types::string::Did;
+use miette::Result;
 use scc::HashSet;
-use serde::{Deserialize, Serialize};
-use smol_str::{SmolStr, ToSmolStr, format_smolstr};
-use std::collections::HashMap;
-use std::ops::{Add, Mul, Sub};
+use smol_str::ToSmolStr;
 use std::sync::Arc;
 use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering};
 use std::time::Duration;
-use tokio::sync::watch;
-use tracing::{Instrument, debug, error, info, trace, warn};
+use tracing::info;
 use url::Url;
 
-const MAX_RETRY_ATTEMPTS: u32 = 5;
-const MAX_RETRY_BATCH: usize = 1000;
+mod by_collection;
+mod relay;
+pub mod throttle;
+mod worker;
 
-#[derive(Debug, Serialize, Deserialize)]
-struct RetryState {
-    after: DateTime<Utc>,
-    duration: TimeDelta,
-    attempts: u32,
-    #[serde(serialize_with = "crate::util::ser_status_code")]
-    #[serde(deserialize_with = "crate::util::deser_status_code")]
-    status: Option<StatusCode>,
-}
+use throttle::Throttler;
 
-impl RetryState {
-    fn new(secs: i64) -> Self {
-        let duration = TimeDelta::seconds(secs);
-        Self {
-            duration,
-            after: Utc::now().add(duration),
-            attempts: 0,
-            status: None,
-        }
-    }
+pub(crate) use by_collection::ByCollectionProducer;
+pub(crate) use relay::{RelayProducer, RetryProducer, SignalChecker};
+pub(crate) use worker::CrawlerWorker;
 
-    /// returns the next retry state with doubled duration and incremented attempt count,
-    /// or `None` if the attempt count would reach the cap (entry left in db as-is).
-    fn next_attempt(self) -> Option<Self> {
-        let attempts = self.attempts + 1;
-        if attempts >= MAX_RETRY_ATTEMPTS {
-            return None;
-        }
-        let duration = self.duration * 2;
-        Some(Self {
-            after: Utc::now().add(duration),
-            duration,
-            attempts,
-            status: None,
-        })
-    }
+// -- InFlight ------------------------------------------------------------
 
-    fn with_status(mut self, code: StatusCode) -> Self {
-        self.status = Some(code);
-        self
-    }
-}
+#[derive(Clone)]
+pub(crate) struct InFlight(Arc<HashSet<Did<'static>>>);
 
-trait ToRetryState {
-    fn to_retry_state(&self) -> RetryState;
-}
-
-impl ToRetryState for ThrottleHandle {
-    fn to_retry_state(&self) -> RetryState {
-        let after = chrono::DateTime::from_timestamp_secs(self.throttled_until()).unwrap();
-        RetryState {
-            duration: after.sub(Utc::now()),
-            after,
-            attempts: 0,
-            status: None,
-        }
+impl InFlight {
+    pub(crate) fn new() -> Self {
+        Self(Arc::new(HashSet::new()))
     }
-}
 
-enum CrawlCheckResult {
-    Signal,
-    NoSignal,
-    Retry(RetryState),
-}
-
-impl From<RetryState> for CrawlCheckResult {
-    fn from(value: RetryState) -> Self {
-        Self::Retry(value)
-    }
-}
-
-fn is_throttle_worthy(e: &reqwest::Error) -> bool {
-    use std::error::Error;
-
-    if e.is_timeout() {
-        return true;
-    }
-
-    let mut src = e.source();
-    while let Some(s) = src {
-        if let Some(io_err) = s.downcast_ref::<std::io::Error>() {
-            if is_tls_cert_error(io_err) {
-                return true;
+    pub(crate) async fn acquire(&self, dids: Vec<Did<'static>>) -> Vec<InFlightGuard> {
+        let mut guards = Vec::with_capacity(dids.len());
+        for did in dids {
+            if self.0.insert_async(did.clone()).await.is_err() {
+                continue;
             }
+            guards.push(InFlightGuard {
+                set: self.clone(),
+                did,
+            });
         }
-        src = s.source();
-    }
-
-    e.status().map_or(false, |s| {
-        matches!(
-            s,
-            StatusCode::BAD_GATEWAY
-                | StatusCode::SERVICE_UNAVAILABLE
-                | StatusCode::GATEWAY_TIMEOUT
-                | crate::util::CONNECTION_TIMEOUT
-                | crate::util::SITE_FROZEN
-        )
-    })
-}
-
-fn is_tls_cert_error(io_err: &std::io::Error) -> bool {
-    let Some(inner) = io_err.get_ref() else {
-        return false;
-    };
-    if let Some(rustls_err) = inner.downcast_ref::<rustls::Error>() {
-        return matches!(rustls_err, rustls::Error::InvalidCertificate(_));
-    }
-    if let Some(nested_io) = inner.downcast_ref::<std::io::Error>() {
-        return is_tls_cert_error(nested_io);
+        guards
     }
-    false
 }
 
-#[derive(Debug, Serialize, Deserialize)]
-enum Cursor {
-    Done(SmolStr),
-    Next(Option<SmolStr>),
+pub(super) struct InFlightGuard {
+    set: InFlight,
+    pub(super) did: Did<'static>,
 }
 
-pub mod throttle;
-use throttle::{OrFailure, Throttler};
-
-type InFlight = Arc<HashSet<Did<'static>>>;
-
-struct InFlightGuard {
-    set: InFlight,
-    did: Did<'static>,
+impl std::ops::Deref for InFlightGuard {
+    type Target = Did<'static>;
+    fn deref(&self) -> &Did<'static> {
+        &self.did
+    }
 }
 
 impl Drop for InFlightGuard {
     fn drop(&mut self) {
-        self.set.remove_sync(&self.did);
+        self.set.0.remove_sync(&self.did);
     }
 }
 
-#[must_use]
-struct InFlightRepos {
-    repos: Vec<Did<'static>>,
-    guards: Vec<InFlightGuard>,
-}
+#[derive(Clone)]
+pub(crate) struct CrawlerStats(Arc<StatsInner>);
 
-pub struct Crawler {
-    state: Arc<AppState>,
-    relays: Vec<Url>,
-    http: reqwest::Client,
-    max_pending: usize,
-    resume_pending: usize,
+struct StatsInner {
+    /// repos committed to the db this interval
     count: AtomicUsize,
+    /// repos seen from relay/by-collection this interval
     crawled_count: AtomicUsize,
     throttled: AtomicBool,
     pds_throttler: Throttler,
-    in_flight: InFlight,
-    enabled: watch::Receiver<bool>,
+    state: Arc<AppState>,
+    relays: Vec<Url>,
 }
 
-impl Crawler {
-    pub fn new(
-        state: Arc<AppState>,
-        relay_hosts: Vec<Url>,
-        max_pending: usize,
-        resume_pending: usize,
-        enabled: watch::Receiver<bool>,
-    ) -> Self {
-        let http = reqwest::Client::builder()
-            .user_agent(concat!(
-                env!("CARGO_PKG_NAME"),
-                "/",
-                env!("CARGO_PKG_VERSION")
-            ))
-            .gzip(true)
-            .build()
-            .expect("that reqwest will build");
-
-        Self {
-            state,
-            relays: relay_hosts,
-            http,
-            max_pending,
-            resume_pending,
+impl CrawlerStats {
+    pub(crate) fn new(state: Arc<AppState>, relays: Vec<Url>, pds_throttler: Throttler) -> Self {
+        Self(Arc::new(StatsInner {
             count: AtomicUsize::new(0),
             crawled_count: AtomicUsize::new(0),
             throttled: AtomicBool::new(false),
-            pds_throttler: Throttler::new(),
-            in_flight: Arc::new(HashSet::new()),
-            enabled,
-        }
+            pds_throttler,
+            state,
+            relays,
+        }))
     }
 
-    async fn get_cursor(&self, relay_host: &Url) -> Result<Cursor> {
-        let key = crawler_cursor_key(relay_host);
-        let cursor_bytes = Db::get(self.state.db.cursors.clone(), &key).await?;
-        let cursor: Cursor = cursor_bytes
-            .as_deref()
-            .map(rmp_serde::from_slice)
-            .transpose()
-            .into_diagnostic()
-            .wrap_err("can't parse cursor")?
-            .unwrap_or(Cursor::Next(None));
-        Ok(cursor)
+    pub(crate) fn record_processed(&self, n: usize) {
+        self.0.count.fetch_add(n, Ordering::Relaxed);
     }
 
-    pub async fn run(self) -> Result<()> {
-        let crawler = Arc::new(self);
-
-        // stats ticker
-        let ticker = tokio::spawn({
-            use std::time::Instant;
-            let crawler = crawler.clone();
-            let mut last_time = Instant::now();
-            let mut interval = tokio::time::interval(Duration::from_secs(60));
-            async move {
-                loop {
-                    interval.tick().await;
-                    let delta_processed = crawler.count.swap(0, Ordering::Relaxed);
-                    let delta_crawled = crawler.crawled_count.swap(0, Ordering::Relaxed);
-                    let is_throttled = crawler.throttled.load(Ordering::Relaxed);
-
-                    crawler.pds_throttler.evict_clean().await;
-
-                    if delta_processed == 0 && delta_crawled == 0 {
-                        if is_throttled {
-                            info!("throttled: pending queue full");
-                        } else {
-                            info!("idle: no repos crawled or processed in 60s");
-                        }
-                        continue;
-                    }
-
-                    let elapsed = last_time.elapsed().as_secs_f64();
-
-                    // fetch all cursors
-                    use futures::future::join_all;
-                    let cursor_futures: Vec<_> = crawler
-                        .relays
-                        .iter()
-                        .map(|relay_host| {
-                            let domain = relay_host.host_str().unwrap_or("unknown");
-                            let relay_host = relay_host.clone();
-                            let crawler = crawler.clone();
-                            async move {
-                                let cursor_str = match crawler.get_cursor(&relay_host).await {
-                                    Ok(c) => match c {
-                                        Cursor::Done(c) => format_smolstr!("done({c})"),
-                                        Cursor::Next(None) => "none".to_smolstr(),
-                                        Cursor::Next(Some(c)) => c.to_smolstr(),
-                                    },
-                                    Err(e) => e.to_smolstr(),
-                                };
-                                format_smolstr!("{domain}={cursor_str}")
-                            }
-                        })
-                        .collect();
-
-                    let cursors: Vec<_> = join_all(cursor_futures).await.into_iter().collect();
-
-                    let cursors_display = if cursors.is_empty() {
-                        "none".to_smolstr()
-                    } else {
-                        cursors.join(", ").into()
-                    };
-
-                    info!(
-                        cursors = %cursors_display,
-                        processed = delta_processed,
-                        crawled = delta_crawled,
-                        elapsed,
-                        "progress"
-                    );
-                    last_time = Instant::now();
-                }
-            }
-        });
-        tokio::spawn(async move {
-            let Err(e) = ticker.await;
-            error!(err = ?e, "stats ticker panicked, aborting");
-            std::process::abort();
-        });
-
-        // retry thread
-        std::thread::spawn({
-            let crawler = crawler.clone();
-            let handle = tokio::runtime::Handle::current();
-            move || {
-                use std::thread::sleep;
-
-                let _g = handle.enter();
-
-                let result = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
-                    loop {
-                        match crawler.process_retry_queue() {
-                            Ok(Some(dur)) => sleep(dur.max(Duration::from_secs(1))),
-                            Ok(None) => sleep(Duration::from_secs(60)),
-                            Err(e) => {
-                                error!(err = %e, "retry loop failed");
-                                sleep(Duration::from_secs(60));
-                            }
-                        }
-                    }
-                }));
-                if result.is_err() {
-                    error!("retry thread panicked, aborting");
-                    std::process::abort();
-                }
-            }
-        });
-
-        info!(
-            relay_count = crawler.relays.len(),
-            hosts = ?crawler.relays,
-            "starting crawler"
-        );
-
-        let mut tasks = tokio::task::JoinSet::new();
-        for url in crawler.relays.clone() {
-            let crawler = crawler.clone();
-            let mut enabled = crawler.enabled.clone();
-            let span = tracing::info_span!("crawl", %url);
-            tasks.spawn(
-                async move {
-                    loop {
-                        if let Err(e) = Self::crawl(crawler.clone(), &url, &mut enabled).await {
-                            error!(err = ?e, "fatal error, restarting in 30s");
-                            tokio::time::sleep(Duration::from_secs(30)).await;
-                        }
-                    }
-                }
-                .instrument(span),
-            );
-        }
-        let _ = tasks.join_all().await;
-
-        Ok(())
+    pub(crate) fn record_crawled(&self, n: usize) {
+        self.0.crawled_count.fetch_add(n, Ordering::Relaxed);
     }
 
-    fn base_url(url: &Url) -> Result<Url> {
-        let mut url = url.clone();
-        match url.scheme() {
-            "wss" => url
-                .set_scheme("https")
-                .map_err(|_| miette::miette!("invalid url: {url}"))?,
-            "ws" => url
-                .set_scheme("http")
-                .map_err(|_| miette::miette!("invalid url: {url}"))?,
-            _ => {}
-        }
-        Ok(url)
+    pub(crate) fn set_throttled(&self, v: bool) {
+        self.0.throttled.store(v, Ordering::Relaxed);
     }
 
-    async fn crawl(
-        crawler: Arc<Self>,
-        relay_host: &Url,
-        enabled: &mut watch::Receiver<bool>,
-    ) -> Result<()> {
-        let base_url = Self::base_url(relay_host)?;
-
-        let mut rng: SmallRng = rand::make_rng();
-        let db = &crawler.state.db;
-
-        let mut cursor = crawler.get_cursor(relay_host).await?;
-
-        match &cursor {
-            Cursor::Next(Some(c)) => info!(cursor = %c, "resuming"),
-            Cursor::Next(None) => info!("starting from scratch"),
-            Cursor::Done(c) => info!(cursor = %c, "was done, resuming"),
-        }
-
-        let mut was_throttled = false;
+    pub(crate) async fn task(self) {
+        use std::time::Instant;
+        let mut last_time = Instant::now();
+        let mut interval = tokio::time::interval(Duration::from_secs(60));
         loop {
-            enabled.wait_enabled("crawler").await;
+            interval.tick().await;
+            let delta_processed = self.0.count.swap(0, Ordering::Relaxed);
+            let delta_crawled = self.0.crawled_count.swap(0, Ordering::Relaxed);
+            let is_throttled = self.0.throttled.load(Ordering::Relaxed);
 
-            // throttle check
-            loop {
-                let pending = crawler.state.db.get_count("pending").await;
-                if pending > crawler.max_pending as u64 {
-                    if !was_throttled {
-                        debug!(
-                            pending,
-                            max = crawler.max_pending,
-                            "throttling: above max pending"
-                        );
-                        was_throttled = true;
-                        crawler.throttled.store(true, Ordering::Relaxed);
-                    }
-                    tokio::select! {
-                        _ = tokio::time::sleep(Duration::from_secs(5)) => {}
-                        _ = enabled.changed() => {
-                            if !*enabled.borrow() { return Ok(()); }
-                        }
-                    }
-                } else if pending > crawler.resume_pending as u64 {
-                    if !was_throttled {
-                        debug!(
-                            pending,
-                            resume = crawler.resume_pending,
-                            "throttling: entering cooldown"
-                        );
-                        was_throttled = true;
-                        crawler.throttled.store(true, Ordering::Relaxed);
-                    }
+            self.0.pds_throttler.evict_clean().await;
 
-                    loop {
-                        let current_pending = crawler.state.db.get_count("pending").await;
-                        if current_pending <= crawler.resume_pending as u64 {
-                            break;
-                        }
-                        debug!(
-                            pending = current_pending,
-                            resume = crawler.resume_pending,
-                            "cooldown, waiting"
-                        );
-                        tokio::select! {
-                            _ = tokio::time::sleep(Duration::from_secs(5)) => {}
-                            _ = enabled.changed() => {
-                                if !*enabled.borrow() { return Ok(()); }
-                            }
-                        }
-                    }
-                    break;
+            if delta_processed == 0 && delta_crawled == 0 {
+                if is_throttled {
+                    info!("throttled: pending queue full");
                 } else {
-                    if was_throttled {
-                        info!("throttling released");
-                        was_throttled = false;
-                        crawler.throttled.store(false, Ordering::Relaxed);
-                    }
-                    break;
+                    info!("idle: no repos crawled or processed in 60s");
                 }
+                continue;
             }
 
-            let mut list_repos_url = base_url
-                .join("/xrpc/com.atproto.sync.listRepos")
-                .into_diagnostic()?;
-            list_repos_url
-                .query_pairs_mut()
-                .append_pair("limit", "1000");
-            if let Cursor::Next(Some(c)) | Cursor::Done(c) = &cursor {
-                list_repos_url
-                    .query_pairs_mut()
-                    .append_pair("cursor", c.as_str());
-            }
+            let elapsed = last_time.elapsed().as_secs_f64();
 
-            let fetch_result = (|| {
-                crawler
-                    .http
-                    .get(list_repos_url.clone())
-                    .send()
-                    .error_for_status()
-            })
-            .retry(5, |e: &reqwest::Error, attempt| {
-                matches!(e.status(), Some(StatusCode::TOO_MANY_REQUESTS))
-                    .then(|| Duration::from_secs(1 << attempt.min(5)))
-            })
-            .await;
-
-            let res = match fetch_result {
-                Ok(r) => r,
-                Err(RetryOutcome::Ratelimited) => {
-                    warn!("rate limited by relay after retries");
-                    continue;
-                }
-                Err(RetryOutcome::Failed(e)) => {
-                    error!(err = %e, "crawler failed to fetch listRepos");
-                    continue;
-                }
-            };
-
-            let bytes = match res.bytes().await {
-                Ok(b) => b,
-                Err(e) => {
-                    error!(err = %e, "cant read listRepos response");
-                    continue;
-                }
-            };
-
-            let mut batch = db.inner.batch();
-            let filter = crawler.state.filter.load();
-
-            struct ParseResult {
-                unknown_dids: Vec<Did<'static>>,
-                cursor: Option<smol_str::SmolStr>,
-                count: usize,
-            }
-
-            const BLOCKING_TASK_TIMEOUT: Duration = Duration::from_secs(30);
-
-            let parse_result = {
-                let repos = db.repos.clone();
-                let filter_ks = db.filter.clone();
-                let crawler_ks = db.crawler.clone();
-
-                // this wont actually cancel the task since spawn_blocking isnt cancel safe
-                // but at least we'll see whats going on?
-                tokio::time::timeout(
-                    BLOCKING_TASK_TIMEOUT,
-                    tokio::task::spawn_blocking(move || -> miette::Result<Option<ParseResult>> {
-                        let output = match serde_json::from_slice::<ListReposOutput>(&bytes) {
-                            Ok(out) => out.into_static(),
-                            Err(e) => {
-                                error!(err = %e, "failed to parse listRepos response");
-                                return Ok(None);
-                            }
-                        };
-
-                        if output.repos.is_empty() {
-                            return Ok(None);
-                        }
-
-                        let count = output.repos.len();
-                        let next_cursor = output.cursor.map(|c| c.as_str().into());
-                        let mut unknown = Vec::new();
-                        for repo in output.repos {
-                            let excl_key = crate::db::filter::exclude_key(repo.did.as_str())?;
-                            if filter_ks.contains_key(&excl_key).into_diagnostic()? {
-                                continue;
-                            }
-
-                            // already in retry queue — let the retry thread handle it
-                            let retry_key = keys::crawler_retry_key(&repo.did);
-                            if crawler_ks.contains_key(&retry_key).into_diagnostic()? {
-                                continue;
-                            }
-
-                            let did_key = keys::repo_key(&repo.did);
-                            if !repos.contains_key(&did_key).into_diagnostic()? {
-                                unknown.push(repo.did.into_static());
-                            }
-                        }
-
-                        Ok(Some(ParseResult {
-                            unknown_dids: unknown,
-                            cursor: next_cursor,
-                            count,
-                        }))
-                    }),
-                )
-                .await
-            }
-            .into_diagnostic()?
-            .map_err(|_| {
-                error!(
-                    "spawn_blocking task for parsing listRepos timed out after {}",
-                    BLOCKING_TASK_TIMEOUT.as_secs()
-                );
-                miette::miette!("spawn_blocking task for parsing listRepos timed out")
-            })?;
-
-            let ParseResult {
-                unknown_dids,
-                cursor: next_cursor,
-                count,
-            } = match parse_result {
-                Ok(Some(res)) => res,
-                Ok(None) => {
-                    info!("finished enumeration (or empty page)");
-                    if let Cursor::Next(Some(c)) = cursor {
-                        info!("reached end of list.");
-                        cursor = Cursor::Done(c);
+            let cursor_futures: Vec<_> = self
+                .0
+                .relays
+                .iter()
+                .map(|relay_host| {
+                    let domain = relay_host.host_str().unwrap_or("unknown").to_owned();
+                    let relay_host = relay_host.clone();
+                    let state = self.0.state.clone();
+                    async move {
+                        let cursor = relay::cursor_display(&state, &relay_host).await;
+                        format!("{domain}={cursor}").into()
                     }
-                    info!("sleeping 1h before next enumeration pass");
-                    tokio::select! {
-                        _ = tokio::time::sleep(Duration::from_secs(3600)) => {
-                            info!("resuming after 1h sleep");
-                        }
-                        _ = enabled.changed() => {
-                            if !*enabled.borrow() { return Ok(()); }
-                        }
-                    }
-                    continue;
-                }
-                Err(e) => return Err(e).wrap_err("error while crawling"),
-            };
-
-            debug!(count, "fetched repos");
-            crawler.crawled_count.fetch_add(count, Ordering::Relaxed);
-
-            let in_flight = if filter.check_signals() && !unknown_dids.is_empty() {
-                // we dont need to pass any existing since we have none; we are crawling after all
-                crawler
-                    .check_signals_batch(&unknown_dids, &filter, &mut batch, &HashMap::new())
-                    .await?
-            } else {
-                // no signal checking but still need dedup to avoid orphan pending entries
-                crawler.acquire_in_flight(unknown_dids).await
-            };
-
-            for did in &in_flight.repos {
-                let did_key = keys::repo_key(did);
-                trace!(did = %did, "found new repo");
-
-                let state = RepoState::untracked(rng.next_u64());
-                batch.insert(&db.repos, &did_key, ser_repo_state(&state)?);
-                batch.insert(&db.pending, keys::pending_key(state.index_id), &did_key);
-            }
-
-            if let Some(new_cursor) = next_cursor {
-                cursor = Cursor::Next(Some(new_cursor.as_str().into()));
-            } else if let Cursor::Next(Some(c)) = cursor {
-                info!("reached end of list.");
-                cursor = Cursor::Done(c);
-            }
-            batch.insert(
-                &db.cursors,
-                crawler_cursor_key(relay_host),
-                rmp_serde::to_vec(&cursor)
-                    .into_diagnostic()
-                    .wrap_err("cant serialize cursor")?,
-            );
-
-            tokio::time::timeout(
-                BLOCKING_TASK_TIMEOUT,
-                tokio::task::spawn_blocking(move || batch.commit().into_diagnostic()),
-            )
-            .await
-            .into_diagnostic()?
-            .map_err(|_| {
-                error!(
-                    "spawn_blocking task for batch commit timed out after {}",
-                    BLOCKING_TASK_TIMEOUT.as_secs()
-                );
-                miette::miette!("spawn_blocking task for batch commit timed out")
-            })?
-            .inspect_err(|e| {
-                error!(err = ?e, "batch commit failed");
-            })
-            .ok();
-
-            drop(in_flight.guards);
-
-            crawler.account_new_repos(in_flight.repos.len()).await;
-
-            if matches!(cursor, Cursor::Done(_)) {
-                info!("enumeration complete, sleeping 1h before next pass");
-                tokio::select! {
-                    _ = tokio::time::sleep(Duration::from_secs(3600)) => {
-                        info!("resuming after 1h sleep");
-                    }
-                    _ = enabled.changed() => {
-                        if !*enabled.borrow() { return Ok(()); }
-                    }
-                }
-            }
-        }
-    }
-
-    fn process_retry_queue(&self) -> Result<Option<Duration>> {
-        let db = &self.state.db;
-        let now = Utc::now();
-
-        let mut ready: Vec<Did> = Vec::new();
-        let mut existing: HashMap<Did<'static>, RetryState> = HashMap::new();
-        let mut next_wake: Option<Duration> = None;
-        let mut had_more = false;
-
-        let mut rng: SmallRng = rand::make_rng();
-
-        let mut batch = db.inner.batch();
-        for guard in db.crawler.prefix(keys::CRAWLER_RETRY_PREFIX) {
-            let (key, val) = guard.into_inner().into_diagnostic()?;
-            let state: RetryState = rmp_serde::from_slice(&val).into_diagnostic()?;
-            let did = keys::crawler_retry_parse_key(&key)?.to_did();
-
-            // leave capped entries alone for API inspection
-            if state.attempts >= MAX_RETRY_ATTEMPTS {
-                continue;
-            }
-
-            let backoff = TimeDelta::seconds(
-                state
-                    .duration
-                    .as_seconds_f64()
-                    .mul(rng.random_range(0.01..0.07)) as i64,
-            );
-            if state.after + backoff > now {
-                let wake = (state.after - now).to_std().unwrap_or(Duration::ZERO);
-                next_wake = Some(next_wake.map(|w| w.min(wake)).unwrap_or(wake));
-                continue;
-            }
-
-            if ready.len() >= MAX_RETRY_BATCH {
-                had_more = true;
-                break;
-            }
-
-            ready.push(did.clone());
-            existing.insert(did, state);
-        }
-
-        if ready.is_empty() {
-            return Ok(next_wake);
-        }
-
-        debug!(count = ready.len(), "retrying pending repos");
-
-        let handle = tokio::runtime::Handle::current();
-        let filter = self.state.filter.load();
-        let in_flight =
-            handle.block_on(self.check_signals_batch(&ready, &filter, &mut batch, &existing))?;
-
-        let mut rng: SmallRng = rand::make_rng();
-        for did in &in_flight.repos {
-            let did_key = keys::repo_key(did);
-
-            if db.repos.contains_key(&did_key).into_diagnostic()? {
-                continue;
-            }
-
-            let state = RepoState::untracked(rng.next_u64());
-            batch.insert(&db.repos, &did_key, ser_repo_state(&state)?);
-            batch.insert(&db.pending, keys::pending_key(state.index_id), &did_key);
-        }
-
-        batch.commit().into_diagnostic()?;
-
-        drop(in_flight.guards);
-
-        if !in_flight.repos.is_empty() {
-            info!(count = in_flight.repos.len(), "recovered from retry queue");
-            handle.block_on(self.account_new_repos(in_flight.repos.len()));
-        }
-
-        // if we hit the batch cap there are more ready entries, loop back immediately
-        Ok(had_more.then_some(Duration::ZERO).or(next_wake))
-    }
-
-    fn check_repo_signals(
-        &self,
-        filter: Arc<crate::filter::FilterConfig>,
-        did: Did<'static>,
-    ) -> impl Future<Output = (Did<'static>, CrawlCheckResult)> + Send + 'static {
-        let resolver = self.state.resolver.clone();
-        let http = self.http.clone();
-        let throttler = self.pds_throttler.clone();
-        async move {
-            const MAX_RETRIES: u32 = 5;
-
-            let pds_url = (|| resolver.resolve_identity_info(&did))
-                .retry(MAX_RETRIES, |e, attempt| {
-                    matches!(e, crate::resolver::ResolverError::Ratelimited)
-                        .then(|| Duration::from_secs(1 << attempt.min(5)))
                 })
-                .await;
-
-            let pds_url = match pds_url {
-                Ok((url, _)) => url,
-                Err(RetryOutcome::Ratelimited) => {
-                    error!(
-                        retries = MAX_RETRIES,
-                        "rate limited resolving identity, giving up"
-                    );
-                    // no pds handle to read retry_after from; use a short default
-                    return (did, RetryState::new(60).into());
-                }
-                Err(RetryOutcome::Failed(e)) => {
-                    error!(err = %e, "failed to resolve identity");
-                    return (did, RetryState::new(60).into());
-                }
-            };
+                .collect();
 
-            let throttle = throttler.get_handle(&pds_url).await;
-            if throttle.is_throttled() {
-                trace!(host = pds_url.host_str(), "skipping throttled pds");
-                return (did, throttle.to_retry_state().into());
-            }
-
-            let _permit = throttle.acquire().unit_error().or_failure(&throttle, || ());
-            let Ok(_permit) = _permit.await else {
-                trace!(
-                    host = pds_url.host_str(),
-                    "pds failed while waiting for permit"
-                );
-                return (did, throttle.to_retry_state().into());
+            let cursors: Vec<smol_str::SmolStr> =
+                join_all(cursor_futures).await.into_iter().collect();
+            let cursors_display = if cursors.is_empty() {
+                "none".to_smolstr()
+            } else {
+                cursors.join(", ").into()
             };
 
-            enum RequestError {
-                Reqwest(reqwest::Error),
-                RateLimited(Option<u64>),
-                /// hard failure notification from another task on this PDS
-                Throttled,
-            }
-
-            let mut describe_url = pds_url.join("/xrpc/com.atproto.repo.describeRepo").unwrap();
-            describe_url.query_pairs_mut().append_pair("repo", &did);
-
-            let resp = async {
-                let resp = http
-                    .get(describe_url)
-                    .timeout(throttle.timeout())
-                    .send()
-                    .await
-                    .map_err(RequestError::Reqwest)?;
-
-                // dont retry ratelimits since we will just put it in a queue to be tried again later
-                if resp.status() == StatusCode::TOO_MANY_REQUESTS {
-                    return Err(RequestError::RateLimited(parse_retry_after(&resp)));
-                }
-
-                resp.error_for_status().map_err(RequestError::Reqwest)
-            }
-            .or_failure(&throttle, || RequestError::Throttled)
-            .await;
-
-            let resp = match resp {
-                Ok(r) => {
-                    throttle.record_success();
-                    r
-                }
-                Err(RequestError::RateLimited(secs)) => {
-                    throttle.record_ratelimit(secs);
-                    return (
-                        did,
-                        throttle
-                            .to_retry_state()
-                            .with_status(StatusCode::TOO_MANY_REQUESTS)
-                            .into(),
-                    );
-                }
-                Err(RequestError::Throttled) => {
-                    return (did, throttle.to_retry_state().into());
-                }
-                Err(RequestError::Reqwest(e)) => {
-                    if e.is_timeout() && !throttle.record_timeout() {
-                        // first or second timeout, just requeue
-                        let mut retry_state = RetryState::new(60);
-                        retry_state.status = e.status();
-                        return (did, retry_state.into());
-                    }
-                    // third timeout, if timeout fail is_throttle_worthy will ban the pds
-
-                    if is_throttle_worthy(&e) {
-                        if let Some(mins) = throttle.record_failure() {
-                            warn!(url = %pds_url, mins, "throttling pds due to hard failure");
-                        }
-                        let mut retry_state = throttle.to_retry_state();
-                        retry_state.status = e.status();
-                        return (did, retry_state.into());
-                    }
-
-                    match e.status() {
-                        Some(StatusCode::NOT_FOUND | StatusCode::GONE) => {
-                            trace!("repo not found");
-                            return (did, CrawlCheckResult::NoSignal);
-                        }
-                        Some(s) if s.is_client_error() => {
-                            error!(status = %s, "repo unavailable");
-                            return (did, CrawlCheckResult::NoSignal);
-                        }
-                        _ => {
-                            error!(err = %e, "repo errored");
-                            let mut retry_state = RetryState::new(60 * 15);
-                            retry_state.status = e.status();
-                            return (did, retry_state.into());
-                        }
-                    }
-                }
-            };
-
-            let bytes = match resp.bytes().await {
-                Ok(b) => b,
-                Err(e) => {
-                    error!(err = %e, "failed to read describeRepo response");
-                    return (did, RetryState::new(60 * 5).into());
-                }
-            };
-
-            let out = match serde_json::from_slice::<DescribeRepoOutput>(&bytes) {
-                Ok(out) => out,
-                Err(e) => {
-                    error!(err = %e, "failed to parse describeRepo response");
-                    return (did, RetryState::new(60 * 10).into());
-                }
-            };
-
-            let found_signal = out
-                .collections
-                .iter()
-                .any(|col| filter.matches_signal(col.as_str()));
-
-            if !found_signal {
-                trace!("no signal-matching collections found");
-            }
-
-            return (
-                did,
-                found_signal
-                    .then_some(CrawlCheckResult::Signal)
-                    .unwrap_or(CrawlCheckResult::NoSignal),
+            info!(
+                cursors = %cursors_display,
+                processed = delta_processed,
+                crawled = delta_crawled,
+                elapsed,
+                "progress"
             );
-        }
-    }
-
-    async fn check_signals_batch(
-        &self,
-        repos: &[Did<'static>],
-        filter: &Arc<crate::filter::FilterConfig>,
-        batch: &mut fjall::OwnedWriteBatch,
-        existing: &HashMap<Did<'static>, RetryState>,
-    ) -> Result<InFlightRepos> {
-        let db = &self.state.db;
-        let in_flight = self.acquire_in_flight(repos.to_vec()).await;
-        let mut valid = Vec::with_capacity(in_flight.repos.len());
-        let mut set = tokio::task::JoinSet::new();
-
-        for did in in_flight.repos {
-            let filter = filter.clone();
-            let span = tracing::info_span!("signals", did = %did);
-            set.spawn(
-                self.check_repo_signals(filter, did.clone())
-                    .instrument(span),
-            );
-        }
-
-        while let Some(res) = tokio::time::timeout(Duration::from_secs(60), set.join_next())
-            .await
-            .into_diagnostic()
-            .map_err(|_| {
-                error!("signal check task timed out after 60s");
-                miette::miette!("signal check task timed out")
-            })?
-        {
-            let (did, result) = match res {
-                Ok(inner) => inner,
-                Err(e) => {
-                    error!(err = ?e, "signal check panicked");
-                    continue;
-                }
-            };
-
-            match result {
-                CrawlCheckResult::Signal => {
-                    batch.remove(&db.crawler, keys::crawler_retry_key(&did));
-                    valid.push(did);
-                }
-                CrawlCheckResult::NoSignal => {
-                    batch.remove(&db.crawler, keys::crawler_retry_key(&did));
-                }
-                CrawlCheckResult::Retry(state) => {
-                    let prev_attempts = existing.get(&did).map(|s| s.attempts).unwrap_or(0);
-                    let carried = RetryState {
-                        attempts: prev_attempts,
-                        ..state
-                    };
-                    let next = match carried.next_attempt() {
-                        Some(next) => next,
-                        None => RetryState {
-                            attempts: MAX_RETRY_ATTEMPTS,
-                            ..state
-                        },
-                    };
-                    batch.insert(
-                        &db.crawler,
-                        keys::crawler_retry_key(&did),
-                        rmp_serde::to_vec(&next).into_diagnostic()?,
-                    );
-                }
-            }
+            last_time = Instant::now();
         }
-
-        Ok(InFlightRepos {
-            repos: valid,
-            guards: in_flight.guards,
-        })
     }
+}
 
-    async fn acquire_in_flight(&self, dids: Vec<Did<'static>>) -> InFlightRepos {
-        let mut filtered = Vec::with_capacity(dids.len());
-        let mut guards = Vec::with_capacity(dids.len());
-        for did in dids {
-            if self.in_flight.insert_async(did.clone()).await.is_err() {
-                trace!(did = %did, "repo in-flight, skipping");
-                continue;
-            }
-            guards.push(InFlightGuard {
-                set: self.in_flight.clone(),
-                did: did.clone(),
-            });
-            filtered.push(did);
-        }
-        InFlightRepos {
-            guards,
-            repos: filtered,
-        }
+/// normalizes a relay URL to http/https for XRPC requests.
+pub(super) fn base_url(url: &Url) -> Result<Url> {
+    let mut url = url.clone();
+    match url.scheme() {
+        "wss" => url
+            .set_scheme("https")
+            .map_err(|_| miette::miette!("invalid url: {url}"))?,
+        "ws" => url
+            .set_scheme("http")
+            .map_err(|_| miette::miette!("invalid url: {url}"))?,
+        _ => {}
     }
-
-    async fn account_new_repos(&self, count: usize) {
-        if count == 0 {
-            return;
-        }
-
-        self.count.fetch_add(count, Ordering::Relaxed);
-        self.state
-            .db
-            .update_count_async("repos", count as i64)
-            .await;
-        self.state
-            .db
-            .update_count_async("pending", count as i64)
-            .await;
-        self.state.notify_backfill();
-    }
+    Ok(url)
 }

src/crawler/relay.rs

···
+use crate::crawler::throttle::{OrFailure, ThrottleHandle, Throttler};
+use crate::db::keys::crawler_cursor_key;
+use crate::db::{Db, keys};
+use crate::state::AppState;
+use crate::util::{
+    ErrorForStatus, RetryOutcome, RetryWithBackoff, WatchEnabledExt, parse_retry_after,
+};
+use chrono::{DateTime, TimeDelta, Utc};
+use fjall::OwnedWriteBatch;
+use futures::FutureExt;
+use jacquard_api::com_atproto::repo::describe_repo::DescribeRepoOutput;
+use jacquard_api::com_atproto::sync::list_repos::ListReposOutput;
+use jacquard_common::{IntoStatic, types::string::Did};
+use miette::{Context, IntoDiagnostic, Result};
+use rand::RngExt;
+use rand::rngs::SmallRng;
+use reqwest::StatusCode;
+use serde::{Deserialize, Serialize};
+use smol_str::{SmolStr, ToSmolStr};
+use std::collections::HashMap;
+use std::ops::{Add, Mul, Sub};
+use std::sync::Arc;
+use std::time::Duration;
+use tokio::sync::{mpsc, watch};
+use tracing::{Instrument, debug, error, info, trace, warn};
+use url::Url;
+
+use super::worker::{CrawlerBatch, CursorUpdate};
+use super::{CrawlerStats, InFlight, InFlightGuard, base_url};
+
+pub(super) const MAX_RETRY_ATTEMPTS: u32 = 5;
+const MAX_RETRY_BATCH: usize = 1000;
+const BLOCKING_TASK_TIMEOUT: Duration = Duration::from_secs(30);
+
+#[derive(Debug, Serialize, Deserialize)]
+pub(super) struct RetryState {
+    pub(super) after: DateTime<Utc>,
+    pub(super) duration: TimeDelta,
+    pub(super) attempts: u32,
+    #[serde(serialize_with = "crate::util::ser_status_code")]
+    #[serde(deserialize_with = "crate::util::deser_status_code")]
+    pub(super) status: Option<StatusCode>,
+}
+
+impl RetryState {
+    pub(super) fn new(secs: i64) -> Self {
+        let duration = TimeDelta::seconds(secs);
+        Self {
+            duration,
+            after: Utc::now().add(duration),
+            attempts: 0,
+            status: None,
+        }
+    }
+
+    /// returns the next retry state or `None` if the attempt count would reach the cap.
+    pub(super) fn next_attempt(self) -> Option<Self> {
+        let attempts = self.attempts + 1;
+        if attempts >= MAX_RETRY_ATTEMPTS {
+            return None;
+        }
+        let duration = self.duration * 2;
+        Some(Self {
+            after: Utc::now().add(duration),
+            duration,
+            attempts,
+            status: None,
+        })
+    }
+
+    pub(super) fn with_status(mut self, code: StatusCode) -> Self {
+        self.status = Some(code);
+        self
+    }
+}
+
+pub(super) enum CrawlCheckResult {
+    Signal,
+    NoSignal,
+    Retry(RetryState),
+}
+
+impl From<RetryState> for CrawlCheckResult {
+    fn from(value: RetryState) -> Self {
+        Self::Retry(value)
+    }
+}
+
+trait ToRetryState {
+    fn to_retry_state(&self) -> RetryState;
+}
+
+impl ToRetryState for ThrottleHandle {
+    fn to_retry_state(&self) -> RetryState {
+        let after = chrono::DateTime::from_timestamp_secs(self.throttled_until()).unwrap();
+        RetryState {
+            duration: after.sub(Utc::now()),
+            after,
+            attempts: 0,
+            status: None,
+        }
+    }
+}
+
+fn is_throttle_worthy(e: &reqwest::Error) -> bool {
+    use std::error::Error;
+
+    if e.is_timeout() {
+        return true;
+    }
+
+    let mut src = e.source();
+    while let Some(s) = src {
+        if let Some(io_err) = s.downcast_ref::<std::io::Error>() {
+            if is_tls_cert_error(io_err) {
+                return true;
+            }
+        }
+        src = s.source();
+    }
+
+    e.status().map_or(false, |s| {
+        matches!(
+            s,
+            StatusCode::BAD_GATEWAY
+                | StatusCode::SERVICE_UNAVAILABLE
+                | StatusCode::GATEWAY_TIMEOUT
+                | crate::util::CONNECTION_TIMEOUT
+                | crate::util::SITE_FROZEN
+        )
+    })
+}
+
+fn is_tls_cert_error(io_err: &std::io::Error) -> bool {
+    let Some(inner) = io_err.get_ref() else {
+        return false;
+    };
+    if let Some(rustls_err) = inner.downcast_ref::<rustls::Error>() {
+        return matches!(rustls_err, rustls::Error::InvalidCertificate(_));
+    }
+    if let Some(nested_io) = inner.downcast_ref::<std::io::Error>() {
+        return is_tls_cert_error(nested_io);
+    }
+    false
+}
+
+// -- SignalChecker --------------------------------------------------------
+
+/// shared describeRepo signal-checking logic used by both relay and retry producers.
+#[derive(Clone)]
+pub(crate) struct SignalChecker {
+    pub(crate) http: reqwest::Client,
+    pub(crate) state: Arc<AppState>,
+    pub(crate) throttler: Throttler,
+}
+
+impl SignalChecker {
+    fn check_repo_signals(
+        &self,
+        filter: Arc<crate::filter::FilterConfig>,
+        did: Did<'static>,
+    ) -> impl Future<Output = (Did<'static>, CrawlCheckResult)> + Send + 'static {
+        let resolver = self.state.resolver.clone();
+        let http = self.http.clone();
+        let throttler = self.throttler.clone();
+        async move {
+            const MAX_RETRIES: u32 = 5;
+
+            let pds_url = (|| resolver.resolve_identity_info(&did))
+                .retry(MAX_RETRIES, |e, attempt| {
+                    matches!(e, crate::resolver::ResolverError::Ratelimited)
+                        .then(|| Duration::from_secs(1 << attempt.min(5)))
+                })
+                .await;
+
+            let pds_url = match pds_url {
+                Ok((url, _)) => url,
+                Err(RetryOutcome::Ratelimited) => {
+                    error!(
+                        retries = MAX_RETRIES,
+                        "rate limited resolving identity, giving up"
+                    );
+                    return (did, RetryState::new(60).into());
+                }
+                Err(RetryOutcome::Failed(e)) => {
+                    error!(err = %e, "failed to resolve identity");
+                    return (did, RetryState::new(60).into());
+                }
+            };
+
+            let throttle = throttler.get_handle(&pds_url).await;
+            if throttle.is_throttled() {
+                trace!(host = pds_url.host_str(), "skipping throttled pds");
+                return (did, throttle.to_retry_state().into());
+            }
+
+            let _permit = throttle.acquire().unit_error().or_failure(&throttle, || ());
+            let Ok(_permit) = _permit.await else {
+                trace!(
+                    host = pds_url.host_str(),
+                    "pds failed while waiting for permit"
+                );
+                return (did, throttle.to_retry_state().into());
+            };
+
+            enum RequestError {
+                Reqwest(reqwest::Error),
+                RateLimited(Option<u64>),
+                Throttled,
+            }
+
+            let mut describe_url = pds_url.join("/xrpc/com.atproto.repo.describeRepo").unwrap();
+            describe_url.query_pairs_mut().append_pair("repo", &did);
+
+            let resp = async {
+                let resp = http
+                    .get(describe_url)
+                    .timeout(throttle.timeout())
+                    .send()
+                    .await
+                    .map_err(RequestError::Reqwest)?;
+                if resp.status() == StatusCode::TOO_MANY_REQUESTS {
+                    return Err(RequestError::RateLimited(parse_retry_after(&resp)));
+                }
+                resp.error_for_status().map_err(RequestError::Reqwest)
+            }
+            .or_failure(&throttle, || RequestError::Throttled)
+            .await;
+
+            let resp = match resp {
+                Ok(r) => {
+                    throttle.record_success();
+                    r
+                }
+                Err(RequestError::RateLimited(secs)) => {
+                    throttle.record_ratelimit(secs);
+                    return (
+                        did,
+                        throttle
+                            .to_retry_state()
+                            .with_status(StatusCode::TOO_MANY_REQUESTS)
+                            .into(),
+                    );
+                }
+                Err(RequestError::Throttled) => {
+                    return (did, throttle.to_retry_state().into());
+                }
+                Err(RequestError::Reqwest(e)) => {
+                    if e.is_timeout() && !throttle.record_timeout() {
+                        let mut retry_state = RetryState::new(60);
+                        retry_state.status = e.status();
+                        return (did, retry_state.into());
+                    }
+                    if is_throttle_worthy(&e) {
+                        if let Some(mins) = throttle.record_failure() {
+                            warn!(url = %pds_url, mins, "throttling pds due to hard failure");
+                        }
+                        let mut retry_state = throttle.to_retry_state();
+                        retry_state.status = e.status();
+                        return (did, retry_state.into());
+                    }
+                    match e.status() {
+                        Some(StatusCode::NOT_FOUND | StatusCode::GONE) => {
+                            trace!("repo not found");
+                            return (did, CrawlCheckResult::NoSignal);
+                        }
+                        Some(s) if s.is_client_error() => {
+                            error!(status = %s, "repo unavailable");
+                            return (did, CrawlCheckResult::NoSignal);
+                        }
+                        _ => {
+                            error!(err = %e, "repo errored");
+                            let mut retry_state = RetryState::new(60 * 15);
+                            retry_state.status = e.status();
+                            return (did, retry_state.into());
+                        }
+                    }
+                }
+            };
+
+            let bytes = match resp.bytes().await {
+                Ok(b) => b,
+                Err(e) => {
+                    error!(err = %e, "failed to read describeRepo response");
+                    return (did, RetryState::new(60 * 5).into());
+                }
+            };
+
+            let out = match serde_json::from_slice::<DescribeRepoOutput>(&bytes) {
+                Ok(out) => out,
+                Err(e) => {
+                    error!(err = %e, "failed to parse describeRepo response");
+                    return (did, RetryState::new(60 * 10).into());
+                }
+            };
+
+            let found_signal = out
+                .collections
+                .iter()
+                .any(|col| filter.matches_signal(col.as_str()));
+            if !found_signal {
+                trace!("no signal-matching collections found");
+            }
+
+            (
+                did,
+                found_signal
+                    .then_some(CrawlCheckResult::Signal)
+                    .unwrap_or(CrawlCheckResult::NoSignal),
+            )
+        }
+    }
+
+    /// checks signals for each DID in `in_flight`, writes retry/no-signal state changes
+    /// into `batch`, and returns a vec containing only the guards for confirmed DIDs.
+    ///
+    /// guards for non-confirmed DIDs are dropped here, releasing their in-flight slots.
+    /// confirmed DIDs' retry entries are NOT removed here — the worker removes them
+    /// atomically with the pending insert.
+    pub(super) async fn check_signals_batch(
+        &self,
+        in_flight: Vec<InFlightGuard>,
+        filter: &Arc<crate::filter::FilterConfig>,
+        batch: &mut OwnedWriteBatch,
+        existing: &HashMap<Did<'static>, RetryState>,
+    ) -> Result<Vec<InFlightGuard>> {
+        let db = &self.state.db;
+        let mut valid: Vec<Did<'static>> = Vec::new();
+        let mut set = tokio::task::JoinSet::new();
+
+        for guard in &in_flight {
+            let filter = filter.clone();
+            let did = guard.did.clone();
+            let span = tracing::info_span!("signals", did = %did);
+            set.spawn(self.check_repo_signals(filter, did).instrument(span));
+        }
+
+        while let Some(res) = tokio::time::timeout(Duration::from_secs(60), set.join_next())
+            .await
+            .into_diagnostic()
+            .map_err(|_| {
+                error!("signal check task timed out after 60s");
+                miette::miette!("signal check task timed out")
+            })?
+        {
+            let (did, result) = match res {
+                Ok(inner) => inner,
+                Err(e) => {
+                    error!(err = ?e, "signal check panicked");
+                    continue;
+                }
+            };
+
+            match result {
+                CrawlCheckResult::Signal => {
+                    valid.push(did);
+                }
+                CrawlCheckResult::NoSignal => {
+                    batch.remove(&db.crawler, keys::crawler_retry_key(&did));
+                }
+                CrawlCheckResult::Retry(state) => {
+                    let prev_attempts = existing.get(&did).map(|s| s.attempts).unwrap_or(0);
+                    let carried = RetryState {
+                        attempts: prev_attempts,
+                        ..state
+                    };
+                    let next = match carried.next_attempt() {
+                        Some(next) => next,
+                        None => RetryState {
+                            attempts: MAX_RETRY_ATTEMPTS,
+                            ..state
+                        },
+                    };
+                    batch.insert(
+                        &db.crawler,
+                        keys::crawler_retry_key(&did),
+                        rmp_serde::to_vec(&next).into_diagnostic()?,
+                    );
+                }
+            }
+        }
+
+        let valid_set: std::collections::HashSet<&Did<'static>> = valid.iter().collect();
+        Ok(in_flight
+            .into_iter()
+            .filter(|g| valid_set.contains(&**g))
+            .collect())
+    }
+}
+
+// -- RelayProducer --------------------------------------------------------
+
+pub(crate) struct RelayProducer {
+    pub(crate) relay_url: Url,
+    pub(crate) checker: SignalChecker,
+    pub(crate) in_flight: InFlight,
+    pub(crate) tx: mpsc::Sender<CrawlerBatch>,
+    pub(crate) enabled: watch::Receiver<bool>,
+    pub(crate) stats: CrawlerStats,
+}
+
+impl RelayProducer {
+    pub(crate) async fn run(mut self) -> Result<()> {
+        loop {
+            if let Err(e) = self.crawl().await {
+                error!(err = ?e, relay = %self.relay_url, "fatal relay crawl error, restarting in 30s");
+                tokio::time::sleep(Duration::from_secs(30)).await;
+            }
+        }
+    }
+
+    async fn get_cursor(&self) -> Result<Option<SmolStr>> {
+        let key = crawler_cursor_key(self.relay_url.as_str());
+        let cursor_bytes = Db::get(self.checker.state.db.cursors.clone(), &key).await?;
+        Ok(cursor_bytes
+            .as_deref()
+            .and_then(|b| rmp_serde::from_slice::<SmolStr>(b).ok()))
+    }
+
+    async fn crawl(&mut self) -> Result<()> {
+        let db = &self.checker.state.db;
+        let base = base_url(&self.relay_url)?;
+
+        let mut cursor = self.get_cursor().await?;
+        match &cursor {
+            Some(c) => info!(cursor = %c, "resuming"),
+            None => info!("starting from scratch"),
+        }
+
+        loop {
+            self.enabled.wait_enabled("crawler").await;
+
+            let mut url = base
+                .join("/xrpc/com.atproto.sync.listRepos")
+                .into_diagnostic()?;
+            url.query_pairs_mut().append_pair("limit", "1000");
+            if let Some(c) = &cursor {
+                url.query_pairs_mut().append_pair("cursor", c.as_str());
+            }
+
+            let fetch_result = (|| self.checker.http.get(url.clone()).send().error_for_status())
+                .retry(5, |e: &reqwest::Error, attempt| {
+                    matches!(e.status(), Some(StatusCode::TOO_MANY_REQUESTS))
+                        .then(|| Duration::from_secs(1 << attempt.min(5)))
+                })
+                .await;
+
+            let res = match fetch_result {
+                Ok(r) => r,
+                Err(RetryOutcome::Ratelimited) => {
+                    warn!("rate limited by relay after retries");
+                    continue;
+                }
+                Err(RetryOutcome::Failed(e)) => {
+                    error!(err = %e, "crawler failed to fetch listRepos");
+                    continue;
+                }
+            };
+
+            let bytes = match res.bytes().await {
+                Ok(b) => b,
+                Err(e) => {
+                    error!(err = %e, "cant read listRepos response");
+                    continue;
+                }
+            };
+
+            let filter = self.checker.state.filter.load();
+
+            struct ParseResult {
+                unknown_dids: Vec<Did<'static>>,
+                cursor: Option<SmolStr>,
+                count: usize,
+            }
+
+            let parse_result = {
+                let repos = db.repos.clone();
+                let filter_ks = db.filter.clone();
+                let crawler_ks = db.crawler.clone();
+
+                tokio::time::timeout(
+                    BLOCKING_TASK_TIMEOUT,
+                    tokio::task::spawn_blocking(move || -> miette::Result<Option<ParseResult>> {
+                        let output = match serde_json::from_slice::<ListReposOutput>(&bytes) {
+                            Ok(out) => out.into_static(),
+                            Err(e) => {
+                                error!(err = %e, "failed to parse listRepos response");
+                                return Ok(None);
+                            }
+                        };
+                        if output.repos.is_empty() {
+                            return Ok(None);
+                        }
+                        let count = output.repos.len();
+                        let next_cursor = output.cursor.map(|c| c.as_str().into());
+                        let mut unknown = Vec::new();
+                        for repo in output.repos {
+                            let excl_key = crate::db::filter::exclude_key(repo.did.as_str())?;
+                            if filter_ks.contains_key(&excl_key).into_diagnostic()? {
+                                continue;
+                            }
+                            let retry_key = keys::crawler_retry_key(&repo.did);
+                            if crawler_ks.contains_key(&retry_key).into_diagnostic()? {
+                                continue;
+                            }
+                            let did_key = keys::repo_key(&repo.did);
+                            if !repos.contains_key(&did_key).into_diagnostic()? {
+                                unknown.push(repo.did.into_static());
+                            }
+                        }
+                        Ok(Some(ParseResult {
+                            unknown_dids: unknown,
+                            cursor: next_cursor,
+                            count,
+                        }))
+                    }),
+                )
+                .await
+            }
+            .into_diagnostic()?
+            .map_err(|_| {
+                error!(
+                    "spawn_blocking task for parsing listRepos timed out after {}s",
+                    BLOCKING_TASK_TIMEOUT.as_secs()
+                );
+                miette::miette!("spawn_blocking task for parsing listRepos timed out")
+            })?;
+
+            let ParseResult {
+                unknown_dids,
+                cursor: next_cursor,
+                count,
+            } = match parse_result {
+                Ok(Some(res)) => res,
+                Ok(None) => {
+                    info!("enumeration pass complete, sleeping 1h");
+                    tokio::select! {
+                        _ = tokio::time::sleep(Duration::from_secs(3600)) => {
+                            info!("resuming after 1h sleep");
+                        }
+                        _ = self.enabled.changed() => {
+                            if !*self.enabled.borrow() { return Ok(()); }
+                        }
+                    }
+                    continue;
+                }
+                Err(e) => return Err(e).wrap_err("error while crawling"),
+            };
+
+            debug!(count, "fetched repos");
+            self.stats.record_crawled(count);
+
+            let pass_complete;
+            if let Some(new_cursor) = next_cursor {
+                cursor = Some(new_cursor.as_str().into());
+                pass_complete = false;
+            } else {
+                info!("reached end of list.");
+                pass_complete = true;
+            }
+
+            let cursor_update = cursor
+                .as_ref()
+                .map(|c| -> Result<CursorUpdate> {
+                    Ok(CursorUpdate {
+                        key: crawler_cursor_key(self.relay_url.as_str()),
+                        value: rmp_serde::to_vec(c.as_str())
+                            .into_diagnostic()
+                            .wrap_err("cant serialize cursor")?,
+                    })
+                })
+                .transpose()?;
+
+            let in_flight = self.in_flight.acquire(unknown_dids).await;
+            let confirmed = if filter.check_signals() && !in_flight.is_empty() {
+                let mut retry_batch = db.inner.batch();
+                let confirmed = self
+                    .checker
+                    .check_signals_batch(in_flight, &filter, &mut retry_batch, &HashMap::new())
+                    .await?;
+                tokio::time::timeout(
+                    BLOCKING_TASK_TIMEOUT,
+                    tokio::task::spawn_blocking(move || retry_batch.commit().into_diagnostic()),
+                )
+                .await
+                .into_diagnostic()?
+                .map_err(|_| miette::miette!("retry state commit timed out"))?
+                .inspect_err(|e| error!(err = ?e, "retry state commit failed"))
+                .ok();
+                confirmed
+            } else {
+                in_flight
+            };
+
+            self.tx
+                .send(CrawlerBatch {
+                    guards: confirmed,
+                    cursor_update,
+                })
+                .await
+                .ok();
+
+            if pass_complete {
+                info!("enumeration complete, sleeping 1h before next pass");
+                tokio::select! {
+                    _ = tokio::time::sleep(Duration::from_secs(3600)) => {
+                        info!("resuming after 1h sleep");
+                    }
+                    _ = self.enabled.changed() => {
+                        if !*self.enabled.borrow() { return Ok(()); }
+                    }
+                }
+            }
+        }
+    }
+}
+
+// -- RelayProducer: cursor display for the stats ticker ------------------
+
+/// formats the current cursor for a relay as a display string, used by the stats ticker.
+pub(super) async fn cursor_display(state: &AppState, relay_host: &Url) -> SmolStr {
+    let key = crawler_cursor_key(relay_host.as_str());
+    let cursor_bytes = match Db::get(state.db.cursors.clone(), &key).await {
+        Ok(b) => b,
+        Err(e) => return e.to_smolstr(),
+    };
+    match cursor_bytes
+        .as_deref()
+        .map(rmp_serde::from_slice::<SmolStr>)
+    {
+        None | Some(Err(_)) => "none".to_smolstr(),
+        Some(Ok(c)) => c,
+    }
+}
+
+// -- RetryProducer --------------------------------------------------------
+
+/// re-checks signal for repos whose `describeRepo` previously failed or timed out.
+pub(crate) struct RetryProducer {
+    pub(crate) checker: SignalChecker,
+    pub(crate) in_flight: InFlight,
+    pub(crate) tx: mpsc::Sender<CrawlerBatch>,
+}
+
+impl RetryProducer {
+    pub(crate) async fn run(self) {
+        loop {
+            match self.process_queue().await {
+                Ok(Some(dur)) => tokio::time::sleep(dur.max(Duration::from_secs(1))).await,
+                Ok(None) => tokio::time::sleep(Duration::from_secs(60)).await,
+                Err(e) => {
+                    error!(err = %e, "retry loop failed");
+                    tokio::time::sleep(Duration::from_secs(60)).await;
+                }
+            }
+        }
+    }
+
+    async fn process_queue(&self) -> Result<Option<Duration>> {
+        let db = self.checker.state.db.clone();
+
+        struct ScanResult {
+            ready: Vec<Did<'static>>,
+            existing: HashMap<Did<'static>, RetryState>,
+            next_wake: Option<Duration>,
+            had_more: bool,
+        }
+
+        let ScanResult {
+            ready,
+            existing,
+            next_wake,
+            had_more,
+        } = tokio::task::spawn_blocking(move || -> Result<ScanResult> {
+            let now = chrono::Utc::now();
+            let mut rng: SmallRng = rand::make_rng();
+            let mut ready: Vec<Did> = Vec::new();
+            let mut existing: HashMap<Did<'static>, RetryState> = HashMap::new();
+            let mut next_wake: Option<Duration> = None;
+            let mut had_more = false;
+
+            for guard in db.crawler.prefix(keys::CRAWLER_RETRY_PREFIX) {
+                let (key, val) = guard.into_inner().into_diagnostic()?;
+                let state: RetryState = rmp_serde::from_slice(&val).into_diagnostic()?;
+                let did = keys::crawler_retry_parse_key(&key)?.to_did();
+
+                if state.attempts >= MAX_RETRY_ATTEMPTS {
+                    continue;
+                }
+
+                let backoff = TimeDelta::seconds(
+                    state
+                        .duration
+                        .as_seconds_f64()
+                        .mul(rng.random_range(0.01..0.07)) as i64,
+                );
+                if state.after + backoff > now {
+                    let wake = (state.after - now).to_std().unwrap_or(Duration::ZERO);
+                    next_wake = Some(next_wake.map(|w| w.min(wake)).unwrap_or(wake));
+                    continue;
+                }
+
+                if ready.len() >= MAX_RETRY_BATCH {
+                    had_more = true;
+                    break;
+                }
+
+                ready.push(did.clone());
+                existing.insert(did, state);
+            }
+
+            Ok(ScanResult {
+                ready,
+                existing,
+                next_wake,
+                had_more,
+            })
+        })
+        .await
+        .into_diagnostic()??;
+
+        if ready.is_empty() {
+            return Ok(next_wake);
+        }
+
+        debug!(count = ready.len(), "retrying pending repos");
+
+        let filter = self.checker.state.filter.load();
+        let in_flight = self.in_flight.acquire(ready).await;
+        let mut retry_batch = self.checker.state.db.inner.batch();
+        let confirmed = self
+            .checker
+            .check_signals_batch(in_flight, &filter, &mut retry_batch, &existing)
+            .await?;
+
+        tokio::task::spawn_blocking(move || retry_batch.commit().into_diagnostic())
+            .await
+            .into_diagnostic()?
+            .inspect_err(|e| error!(err = ?e, "retry state commit failed"))
+            .ok();
+
+        if !confirmed.is_empty() {
+            info!(count = confirmed.len(), "recovered from retry queue");
+            self.tx
+                .send(CrawlerBatch {
+                    guards: confirmed,
+                    cursor_update: None,
+                })
+                .await
+                .ok();
+        }
+
+        Ok(had_more.then_some(Duration::ZERO).or(next_wake))
+    }
+}

src/crawler/worker.rs

···
+use crate::db::{keys, ser_repo_state};
+use crate::state::AppState;
+use crate::types::RepoState;
+use miette::{IntoDiagnostic, Result};
+use rand::Rng;
+use rand::rngs::SmallRng;
+use std::sync::Arc;
+use std::time::Duration;
+use tokio::sync::mpsc;
+use tracing::{debug, error, info, trace};
+
+use super::{CrawlerStats, InFlightGuard};
+
+const WORKER_CHANNEL_CAPACITY: usize = 64;
+const BLOCKING_TASK_TIMEOUT: Duration = Duration::from_secs(30);
+
+/// a cursor write to include atomically with the batch.
+pub(crate) struct CursorUpdate {
+    pub(super) key: Vec<u8>,
+    pub(super) value: Vec<u8>,
+}
+
+/// a batch of confirmed repos from any crawler source ready to enqueue.
+///
+/// `guards` hold the DIDs (via `Deref`) and keep their in-flight slots occupied
+/// until the batch is committed to the database.  `cursor_update`, if present,
+/// is committed atomically with the repos/pending inserts.
+pub(crate) struct CrawlerBatch {
+    pub(super) guards: Vec<InFlightGuard>,
+    pub(super) cursor_update: Option<CursorUpdate>,
+}
+
+pub(crate) struct CrawlerWorker {
+    pub(super) state: Arc<AppState>,
+    pub(super) max_pending: usize,
+    pub(super) resume_pending: usize,
+    pub(super) stats: CrawlerStats,
+    pub(super) rx: mpsc::Receiver<CrawlerBatch>,
+    pub(super) was_throttled: bool,
+}
+
+impl CrawlerWorker {
+    pub(crate) fn new(
+        state: Arc<AppState>,
+        max_pending: usize,
+        resume_pending: usize,
+        stats: CrawlerStats,
+    ) -> (Self, mpsc::Sender<CrawlerBatch>) {
+        let (tx, rx) = mpsc::channel(WORKER_CHANNEL_CAPACITY);
+        (
+            Self {
+                state,
+                max_pending,
+                resume_pending,
+                stats,
+                rx,
+                was_throttled: false,
+            },
+            tx,
+        )
+    }
+
+    pub(crate) async fn run(mut self) {
+        while let Some(batch) = self.rx.recv().await {
+            self.wait_for_capacity().await;
+            if let Err(e) = self.enqueue(batch).await {
+                error!(err = ?e, "crawler worker: enqueue failed");
+            }
+        }
+    }
+
+    /// blocks until the pending queue has capacity.  mirrors the hysteresis logic
+    /// that was previously duplicated in each crawler loop:
+    ///   - above `max_pending`: hard stop, poll every 5s
+    ///   - between `resume_pending` and `max_pending`: cooldown until below `resume_pending`
+    ///   - below `resume_pending`: proceed (or release throttle)
+    async fn wait_for_capacity(&mut self) {
+        loop {
+            let pending = self.state.db.get_count("pending").await;
+            if pending > self.max_pending as u64 {
+                if !self.was_throttled {
+                    debug!(
+                        pending,
+                        max = self.max_pending,
+                        "throttling: above max pending"
+                    );
+                    self.was_throttled = true;
+                    self.stats.set_throttled(true);
+                }
+                tokio::time::sleep(Duration::from_secs(5)).await;
+            } else if pending > self.resume_pending as u64 {
+                if !self.was_throttled {
+                    debug!(
+                        pending,
+                        resume = self.resume_pending,
+                        "throttling: entering cooldown"
+                    );
+                    self.was_throttled = true;
+                    self.stats.set_throttled(true);
+                }
+                loop {
+                    tokio::time::sleep(Duration::from_secs(5)).await;
+                    if self.state.db.get_count("pending").await <= self.resume_pending as u64 {
+                        break;
+                    }
+                }
+                self.was_throttled = false;
+                self.stats.set_throttled(false);
+                info!("throttling released");
+                break;
+            } else {
+                if self.was_throttled {
+                    self.was_throttled = false;
+                    self.stats.set_throttled(false);
+                    info!("throttling released");
+                }
+                break;
+            }
+        }
+    }
+
+    /// filters already-known repos, commits them to `repos` + `pending` (plus
+    /// an optional cursor update), then releases the in-flight guards.
+    async fn enqueue(&mut self, batch: CrawlerBatch) -> Result<()> {
+        let CrawlerBatch {
+            guards,
+            cursor_update,
+        } = batch;
+
+        // nothing to insert but still need to commit the cursor if present.
+        if guards.is_empty() {
+            if let Some(cursor) = cursor_update {
+                self.commit_cursor(cursor).await?;
+            }
+            return Ok(());
+        }
+
+        // filter already-known repos, build and commit the write batch, then return
+        // the surviving guards so they are dropped on the async side after commit.
+        let db = self.state.db.clone();
+        let surviving = tokio::time::timeout(
+            BLOCKING_TASK_TIMEOUT,
+            tokio::task::spawn_blocking(move || -> Result<Vec<InFlightGuard>> {
+                let mut rng: SmallRng = rand::make_rng();
+                let mut write_batch = db.inner.batch();
+                let mut surviving = Vec::new();
+                for guard in guards {
+                    let did_key = keys::repo_key(&*guard);
+                    if db.repos.contains_key(&did_key).into_diagnostic()? {
+                        continue;
+                    }
+                    let state = RepoState::untracked(rng.next_u64());
+                    write_batch.insert(&db.repos, &did_key, ser_repo_state(&state)?);
+                    write_batch.insert(&db.pending, keys::pending_key(state.index_id), &did_key);
+                    // clear any stale retry entry, this DID is confirmed and being enqueued
+                    write_batch.remove(&db.crawler, keys::crawler_retry_key(&*guard));
+                    trace!(did = %*guard, "enqueuing repo");
+                    surviving.push(guard);
+                }
+                if let Some(cursor) = cursor_update {
+                    write_batch.insert(&db.cursors, cursor.key, cursor.value);
+                }
+                write_batch.commit().into_diagnostic()?;
+                Ok(surviving)
+            }),
+        )
+        .await
+        .into_diagnostic()?
+        .map_err(|_| {
+            error!("enqueue batch timed out after {BLOCKING_TASK_TIMEOUT:?}");
+            miette::miette!("enqueue batch timed out")
+        })?
+        .inspect_err(|e| error!(err = ?e, "enqueue batch commit failed"))
+        .unwrap_or_default();
+
+        let count = surviving.len();
+        // release in-flight slots now that the batch is committed
+        drop(surviving);
+
+        if count > 0 {
+            self.stats.record_processed(count);
+            self.state
+                .db
+                .update_count_async("repos", count as i64)
+                .await;
+            self.state
+                .db
+                .update_count_async("pending", count as i64)
+                .await;
+            self.state.notify_backfill();
+        }
+
+        Ok(())
+    }
+
+    async fn commit_cursor(&self, cursor: CursorUpdate) -> Result<()> {
+        let db = self.state.db.clone();
+        tokio::time::timeout(
+            BLOCKING_TASK_TIMEOUT,
+            tokio::task::spawn_blocking(move || {
+                let mut batch = db.inner.batch();
+                batch.insert(&db.cursors, cursor.key, cursor.value);
+                batch.commit().into_diagnostic()
+            }),
+        )
+        .await
+        .into_diagnostic()?
+        .map_err(|_| miette::miette!("cursor-only commit timed out"))?
+    }
+}

src/db/keys.rs

···
 use smol_str::SmolStr;
 
 use crate::db::types::{DbRkey, DbTid, TrimmedDid};
-use url::Url;
 
 /// separator used for composite keys
 pub const SEP: u8 = b'|';
···
     TrimmedDid::try_from(&key[CRAWLER_RETRY_PREFIX.len()..])
 }
 
-pub fn crawler_cursor_key(relay: &Url) -> Vec<u8> {
+pub fn crawler_cursor_key(relay: &str) -> Vec<u8> {
     let mut key = b"crawler_cursor|".to_vec();
-    key.extend_from_slice(relay.as_str().as_bytes());
+    key.extend_from_slice(relay.as_bytes());
     key
 }
 
-pub fn firehose_cursor_key(relay: &Url) -> Vec<u8> {
+pub fn by_collection_cursor_key(url: &str, collection: &str) -> Vec<u8> {
+    let mut key = b"by_collection_cursor|".to_vec();
+    key.extend_from_slice(url.as_bytes());
+    key.push(SEP);
+    key.extend_from_slice(collection.as_bytes());
+    key
+}
+
+/// prefix for all by-collection cursors belonging to a given index URL.
+pub fn by_collection_cursor_prefix(url: &str) -> Vec<u8> {
+    let mut prefix = b"by_collection_cursor|".to_vec();
+    prefix.extend_from_slice(url.as_bytes());
+    prefix.push(SEP);
+    prefix
+}
+
+pub fn firehose_cursor_key(relay: &str) -> Vec<u8> {
     let mut key = b"firehose_cursor|".to_vec();
-    key.extend_from_slice(relay.as_str().as_bytes());
+    key.extend_from_slice(relay.as_bytes());
     key
 }
 

src/db/mod.rs

···
             .max_journaling_size(mb(cfg.db_max_journaling_size_mb))
             .with_compaction_filter_factories({
                 let ephemeral = cfg.ephemeral;
-                let f = move |ks: &str| {
-                    tracing::info!("with_compaction_filter_factories queried for keyspace: {ks}",);
-                    match ks {
-                        "counts" => ephemeral.then(|| -> Arc<dyn Factory> {
-                            Arc::new(DropPrefixFilterFactory {
-                                prefix: keys::COUNT_COLLECTION_PREFIX,
-                            })
-                        }),
-                        _ => None,
-                    }
+                let f = move |ks: &str| match ks {
+                    "counts" => ephemeral.then(|| -> Arc<dyn Factory> {
+                        Arc::new(DropPrefixFilterFactory {
+                            prefix: keys::COUNT_COLLECTION_PREFIX,
+                        })
+                    }),
+                    _ => None,
                 };
                 Arc::new(f)
             })
···
             let path = cfg.database_path.join(format!("dict_{name}.bin"));
             if path.exists() {
                 if let Ok(bytes) = std::fs::read(&path) {
-                    tracing::info!(
+                    tracing::debug!(
                         "loaded zstd dictionary for keyspace {name} ({} bytes)",
                         bytes.len()
                     );
···
 
 pub fn set_firehose_cursor(db: &Db, relay: &Url, cursor: i64) -> Result<()> {
     db.cursors
-        .insert(keys::firehose_cursor_key(relay), cursor.to_be_bytes())
+        .insert(
+            keys::firehose_cursor_key(relay.as_str()),
+            cursor.to_be_bytes(),
+        )
         .into_diagnostic()
 }
 
 pub async fn get_firehose_cursor(db: &Db, relay: &Url) -> Result<Option<i64>> {
-    let per_relay_key = keys::firehose_cursor_key(relay);
+    let per_relay_key = keys::firehose_cursor_key(relay.as_str());
     if let Some(v) = Db::get(db.cursors.clone(), per_relay_key).await? {
         return Ok(Some(i64::from_be_bytes(
             v.as_ref()

src/state.rs

···
 
         let crawler_default = match config.enable_crawler {
             Some(b) => b,
-            // default: enabled in full-network mode, disabled in filter mode
-            None => filter_config.mode == crate::filter::FilterMode::Full,
+            // default: enabled if full-network mode, or if crawler sources are configured
+            None => {
+                filter_config.mode == crate::filter::FilterMode::Full
+                    || !config.crawler_sources.is_empty()
+            }
         };
 
         let filter = new_handle(filter_config);

tests/collection_index_test.nu

···
+#!/usr/bin/env nu
+# tests that the collection-index crawler (listReposByCollection) correctly discovers
+# and backfills repos from a lightrail-style index server.
+#
+# usage: nu tests/collection_index_test.nu
+#
+# requires network access to lightrail.microcosm.blue.
+use common.nu *
+
+def main [] {
+    let port = 3015
+    let url = $"http://localhost:($port)"
+    let db_path = (mktemp -d -t hydrant_collection_index_test.XXXXXX)
+    let collection = "app.bsky.graph.starterpack"
+    let index_url = "https://lightrail.microcosm.blue"
+
+    print $"database path: ($db_path)"
+
+    # fetch a small known set of repos from the collection index so we can verify
+    # they appear in hydrant after the discovery pass
+    print $"fetching known repos for ($collection) from ($index_url)..."
+    let index_resp = (http get $"($index_url)/xrpc/com.atproto.sync.listReposByCollection?collection=($collection)&limit=3")
+    let known_dids = ($index_resp.repos | each { |r| $r.did })
+
+    if ($known_dids | is-empty) {
+        print "SKIP: collection index returned no repos, cannot verify discovery"
+        rm -rf $db_path
+        exit 0
+    }
+
+    print $"will verify these DIDs are discovered: ($known_dids | str join ', ')"
+
+    # start hydrant in filter mode with the test collection as a signal.
+    # HYDRANT_ENABLE_COLLECTION_INDEX is true by default when signals are set,
+    # so no explicit override needed.
+    $env.HYDRANT_FILTER_SIGNALS = $collection
+    $env.HYDRANT_COLLECTION_INDEX_URL = $index_url
+    # keep the pending queue very small so throttling doesn't interfere with the test
+    $env.HYDRANT_CRAWLER_MAX_PENDING_REPOS = "500"
+    $env.HYDRANT_CRAWLER_RESUME_PENDING_REPOS = "200"
+    # no relay needed for collection-index testing
+    $env.HYDRANT_ENABLE_FIREHOSE = "false"
+
+    let binary = build-hydrant
+    let instance = start-hydrant $binary $db_path $port
+
+    mut test_passed = false
+
+    if not (wait-for-api $url) {
+        print "ERROR: hydrant failed to start"
+        try { kill -9 $instance.pid }
+        rm -rf $db_path
+        exit 1
+    }
+
+    # verify the filter was applied correctly
+    let filter = (http get $"($url)/filter")
+    print $"filter state: ($filter | to json)"
+    if not ($filter.signals | any { |s| $s == $collection }) {
+        print $"FAILED: ($collection) not in signals — filter not configured"
+        try { kill -9 $instance.pid }
+        rm -rf $db_path
+        exit 1
+    }
+
+    # wait for the collection-index pass to discover and enqueue repos.
+    # the first pass runs immediately on startup; repos should appear within ~30s
+    # depending on pagination speed and backfill concurrency.
+    print "waiting for collection-index discovery pass..."
+    mut discovered = false
+    for i in 1..60 {
+        let stats = (try { (http get $"($url)/stats?accurate=true").counts } catch { {} })
+        let repos = ($stats | get --optional repos | default 0 | into int)
+        print $"[($i)/60] repos: ($repos)"
+        if $repos >= ($known_dids | length) {
+            $discovered = true
+            break
+        }
+        sleep 2sec
+    }
+
+    if not $discovered {
+        print "FAILED: collection-index did not discover any repos within timeout"
+        try { kill -9 $instance.pid }
+        rm -rf $db_path
+        exit 1
+    }
+
+    # verify each known DID appears in the repos API
+    print "verifying known DIDs were discovered..."
+    mut all_found = true
+    for did in $known_dids {
+        let repo = (try {
+            http get $"($url)/repos/($did)"
+        } catch {
+            null
+        })
+        if ($repo | is-empty) {
+            print $"FAILED: ($did) not found in repos API"
+            $all_found = false
+        } else {
+            print $"ok: ($did) — status: ($repo.status)"
+        }
+    }
+
+    if $all_found {
+        print "test PASSED: collection-index correctly discovered repos from listReposByCollection"
+        $test_passed = true
+    }
+
+    print "stopping hydrant..."
+    try { kill -9 $instance.pid }
+    rm -rf $db_path
+
+    if $test_passed {
+        exit 0
+    } else {
+        exit 1
+    }
+}