diff --git a/src/crawler/mod.rs b/src/crawler/mod.rs index 607ba6b..4c71c58 100644 --- a/src/crawler/mod.rs +++ b/src/crawler/mod.rs @@ -14,6 +14,7 @@ use url::Url; mod by_collection; mod list_repos; +mod pds_repos; mod relay_first; mod worker; diff --git a/src/crawler/pds_repos.rs b/src/crawler/pds_repos.rs new file mode 100644 index 0000000..e5524d7 --- /dev/null +++ b/src/crawler/pds_repos.rs @@ -0,0 +1,322 @@ +use std::collections::HashMap; +use std::time::Duration; + +use jacquard_api::com_atproto::sync::list_repos::ListReposOutput; +use jacquard_common::{IntoStatic, types::string::Did}; +use miette::{Context, IntoDiagnostic, Result}; +use reqwest::StatusCode; +use smol_str::SmolStr; +use tokio::sync::{mpsc, watch}; +use tracing::debug; +use url::Url; + +use super::worker::{CrawlerBatch, CursorUpdate}; +use super::{CrawlerStats, InFlight, SignalChecker}; +use crate::db::keys; +use crate::net::{PublicEndpoint, PublicHttpClient}; +use crate::util::{WatchEnabledExt, parse_retry_after}; + +const MAX_LIST_RESPONSE_BYTES: usize = 10 * 1024 * 1024; +const BLOCKING_TASK_TIMEOUT: Duration = Duration::from_secs(30); +const MAX_PAGES: usize = 100_000; + +/// how a walk over one PDS ended. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub(crate) enum PdsListing { + Complete, + /// the host was banned locally partway through, so the rest was skipped. + Banned, +} + +/// walks one PDS's `listRepos` and hands the repos we don't know yet to the crawler worker. +#[derive(Clone)] +pub(crate) struct PdsRepoLister { + pub(crate) checker: SignalChecker, + pub(crate) in_flight: InFlight, + pub(crate) tx: mpsc::Sender, + pub(crate) enabled: watch::Receiver, + pub(crate) stats: CrawlerStats, + pub(crate) component: &'static str, +} + +impl PdsRepoLister { + /// `progress` gets the cursor of the next page, or `None` once the walk is done, and + /// returns the cursor writes to commit together with the page that produced it. + pub(crate) async fn list( + &mut self, + pds: &Url, + account_count: u64, + http: &PublicHttpClient, + mut cursor: Option, + progress: impl Fn(Option<&SmolStr>) -> Result>, + ) -> Result { + let hostname = pds + .host_str() + .ok_or_else(|| miette::miette!("PDS URL has no hostname"))?; + let mut page_count = 0usize; + + loop { + page_count += 1; + if page_count > MAX_PAGES { + miette::bail!("PDS listRepos exceeded {MAX_PAGES} pages"); + } + self.enabled.wait_enabled(self.component).await; + if self.checker.state.pds_meta.load().is_banned(hostname) { + debug!(%hostname, "locally banned PDS, stopping listRepos"); + return Ok(PdsListing::Banned); + } + let mut url = pds + .join("/xrpc/com.atproto.sync.listRepos") + .into_diagnostic()?; + url.query_pairs_mut().append_pair("limit", "1000"); + if let Some(value) = &cursor { + url.query_pairs_mut().append_pair("cursor", value); + } + + let Some(bytes) = self + .fetch_page(pds, hostname, account_count, url, http) + .await? + else { + self.send_cursor_updates(progress(None)?).await?; + return Ok(PdsListing::Complete); + }; + let page = self.parse_page(bytes).await?; + self.stats.record_crawled(page.count); + + let in_flight = self.in_flight.acquire(page.unknown_dids).await; + let filter = self.checker.state.filter.load_full(); + let confirmed = if filter.check_signals() && !in_flight.is_empty() { + let mut retry_batch = self.checker.state.db.inner.batch(); + let confirmed = self + .checker + .check_signals_batch(in_flight, &filter, &mut retry_batch, &HashMap::new()) + .await?; + self.checker + .state + .db + .run(move |_db| retry_batch.commit().into_diagnostic()) + .await?; + confirmed + } else { + in_flight + }; + + let next_cursor = page.cursor; + if next_cursor.is_some() && next_cursor == cursor { + miette::bail!("PDS listRepos returned the same cursor twice"); + } + self.tx + .send(CrawlerBatch { + guards: confirmed, + listings: Vec::new(), + cursor_updates: progress(next_cursor.as_ref())?, + }) + .await + .into_diagnostic() + .wrap_err("crawler worker exited")?; + + let Some(next) = next_cursor else { + return Ok(PdsListing::Complete); + }; + cursor = Some(next); + } + } + + async fn fetch_page( + &self, + pds: &Url, + hostname: &str, + account_count: u64, + url: Url, + http: &PublicHttpClient, + ) -> Result> { + let throttle = self.checker.throttler.get_handle(pds).await; + if throttle.is_throttled() { + miette::bail!("PDS is throttled until {}", throttle.throttled_until()); + } + let _permit = throttle.acquire().await; + let tier = self.checker.state.resolve_pds_tier(hostname); + throttle.wait_for_allow(account_count, &tier).await; + if throttle.is_throttled() { + miette::bail!("PDS became throttled while waiting for pacing"); + } + + let endpoint = PublicEndpoint::parse_http(&url) + .into_diagnostic() + .wrap_err("unsafe PDS endpoint for listRepos")?; + let response = match http.get(&endpoint).timeout(throttle.timeout()).send().await { + Ok(response) => response, + Err(err) => { + throttle.record_failure_detail("list_repos", err.to_string()); + return Err(err).into_diagnostic(); + } + }; + if response.status() == StatusCode::TOO_MANY_REQUESTS { + throttle.record_ratelimit(parse_retry_after(&response)); + miette::bail!("PDS rate limited listRepos"); + } + if matches!( + response.status(), + StatusCode::BAD_REQUEST | StatusCode::NOT_FOUND | StatusCode::METHOD_NOT_ALLOWED + ) || response.status().is_redirection() + { + debug!(status = %response.status(), "PDS does not support listRepos; skipping"); + throttle.record_success(); + return Ok(None); + } + let response = match response.error_for_status() { + Ok(response) => response, + Err(err) => { + throttle.record_failure_detail("list_repos", err.to_string()); + return Err(err).into_diagnostic(); + } + }; + let bytes = crate::util::read_limited_bytes(response, MAX_LIST_RESPONSE_BYTES).await?; + throttle.record_success(); + Ok(Some(bytes)) + } + + async fn parse_page(&self, bytes: bytes::Bytes) -> Result { + let repos = self.checker.state.db.repos.clone(); + let filter = self.checker.state.db.filter.clone(); + let crawler = self.checker.state.db.crawler.clone(); + tokio::time::timeout( + BLOCKING_TASK_TIMEOUT, + tokio::task::spawn_blocking(move || { + let output = serde_json::from_slice::(&bytes) + .into_diagnostic() + .wrap_err("failed to parse PDS listRepos response")? + .into_static(); + select_unknown_repos(&repos, &filter, &crawler, output) + }), + ) + .await + .map_err(|_| miette::miette!("PDS listRepos parse timed out"))? + .into_diagnostic()? + } + + pub(super) async fn send_cursor_updates( + &self, + cursor_updates: Vec, + ) -> Result<()> { + self.tx + .send(CrawlerBatch { + guards: Vec::new(), + listings: Vec::new(), + cursor_updates, + }) + .await + .into_diagnostic() + .wrap_err("crawler worker exited") + } +} + +struct Page { + unknown_dids: Vec, + cursor: Option, + count: usize, +} + +fn select_unknown_repos( + repos: &fjall::Keyspace, + filter: &fjall::Keyspace, + crawler: &fjall::Keyspace, + output: ListReposOutput, +) -> Result { + let count = output.repos.len(); + let cursor = output.cursor; + let mut unknown_dids = Vec::new(); + for repo in output.repos { + let exclude_key = crate::db::filter::exclude_key(repo.did.as_str())?; + let did_key = keys::repo_key(&repo.did); + let retry_key = keys::crawler_retry_key(&repo.did); + let known = filter.contains_key(exclude_key).into_diagnostic()? + || repos.contains_key(did_key).into_diagnostic()? + || crawler.contains_key(retry_key).into_diagnostic()?; + if !known { + unknown_dids.push(repo.did); + } + } + Ok(Page { + unknown_dids, + cursor, + count, + }) +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::config::Config; + use crate::state::AppState; + use serde_json::json; + use tempfile::tempdir; + + const HEAD: &str = "bafyreidf747c4x3lps3k4n357l3a3r57k3k465743k573k465743k5"; + + fn list_repos_output(value: serde_json::Value) -> Result { + let bytes = serde_json::to_vec(&value).into_diagnostic()?; + serde_json::from_slice::(&bytes).into_diagnostic() + } + + #[test] + fn pages_select_only_unknown_repos_without_trusting_status() -> Result<()> { + let tmp = tempdir().into_diagnostic()?; + let config = Config { + database_path: tmp.path().to_path_buf(), + ..Default::default() + }; + let state = AppState::new(&config)?; + let known: Did = Did::new_static("did:web:known.example")?; + let excluded: Did = Did::new_static("did:web:excluded.example")?; + let retry: Did = Did::new_static("did:web:retry.example")?; + state + .db + .repos + .insert(keys::repo_key(&known), b"relay") + .into_diagnostic()?; + state + .db + .filter + .insert(crate::db::filter::exclude_key(excluded.as_str())?, b"") + .into_diagnostic()?; + state + .db + .crawler + .insert(keys::crawler_retry_key(&retry), b"") + .into_diagnostic()?; + + let page = select_unknown_repos( + &state.db.repos, + &state.db.filter, + &state.db.crawler, + list_repos_output(json!({ + "cursor": "next", + "repos": [ + {"did": known, "head": HEAD, "rev": "3j6s746574657", "active": false, "status": "takendown"}, + {"did": excluded, "head": HEAD, "rev": "3j6s746574657"}, + {"did": retry, "head": HEAD, "rev": "3j6s746574657"}, + {"did": "did:web:inactive-unknown.example", "head": HEAD, "rev": "3j6s746574657", "active": false, "status": "deactivated"} + ] + }))?, + )?; + + assert_eq!(page.count, 4); + assert_eq!(page.cursor.as_deref(), Some("next")); + assert_eq!(page.unknown_dids.len(), 1); + assert_eq!( + page.unknown_dids[0].as_str(), + "did:web:inactive-unknown.example" + ); + assert_eq!( + state + .db + .repos + .get(keys::repo_key(&known)) + .into_diagnostic()? + .as_deref(), + Some(b"relay".as_slice()) + ); + Ok(()) + } +} diff --git a/src/crawler/relay_first.rs b/src/crawler/relay_first.rs index d2b4450..3516d16 100644 --- a/src/crawler/relay_first.rs +++ b/src/crawler/relay_first.rs @@ -4,10 +4,8 @@ use std::time::Duration; use jacquard_api::com_atproto::sync::HostStatus; use jacquard_api::com_atproto::sync::list_hosts::{Host, ListHostsOutput}; -use jacquard_api::com_atproto::sync::list_repos::ListReposOutput; -use jacquard_common::{IntoStatic, types::string::Did}; +use jacquard_common::types::string::Did; use miette::{Context, IntoDiagnostic, Result}; -use reqwest::StatusCode; use serde::{Deserialize, Serialize}; use smol_str::SmolStr; use tokio::sync::{mpsc, watch}; @@ -15,21 +13,20 @@ use tracing::{debug, error, info, trace, warn}; use url::Url; use super::list_repos::ListReposProducer; +use super::pds_repos::PdsRepoLister; use super::worker::{CrawlerBatch, CursorUpdate}; use super::{CrawlerStats, InFlight, SignalChecker, base_url}; use crate::db::keys; -use crate::net::{PublicEndpoint, PublicHost, PublicHttpClient, public_http, trusted_http}; +use crate::net::{PublicHost, PublicHttpClient, public_http, trusted_http}; use crate::pds_discovery; use crate::state::AppState; -use crate::util::{WatchEnabledExt, parse_retry_after}; +use crate::util::WatchEnabledExt; const MAX_CONCURRENT_PDS_CRAWLS: usize = 4; const MAX_LIST_RESPONSE_BYTES: usize = 10 * 1024 * 1024; const PASS_POLL_INTERVAL: Duration = Duration::from_secs(1); const RETRY_DELAY: Duration = Duration::from_secs(30); -const BLOCKING_TASK_TIMEOUT: Duration = Duration::from_secs(30); const MAX_LIST_HOSTS_PAGES: usize = 100; -const MAX_PDS_PAGES: usize = 100_000; #[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)] struct PassCursor { @@ -80,11 +77,14 @@ impl RelayFirstProducer { }; let deep = DeepDiscoveryProducer { relay_url: self.url, - checker: self.checker, - in_flight: self.in_flight, - tx: self.tx, - enabled: self.enabled, - stats: self.stats, + lister: PdsRepoLister { + checker: self.checker, + in_flight: self.in_flight, + tx: self.tx, + enabled: self.enabled, + stats: self.stats, + component: "relay-first deep crawler", + }, }; tokio::select! { @@ -94,19 +94,19 @@ impl RelayFirstProducer { } } +#[derive(Clone)] struct DeepDiscoveryProducer { relay_url: Url, - checker: SignalChecker, - in_flight: InFlight, - tx: mpsc::Sender, - enabled: watch::Receiver, - stats: CrawlerStats, + lister: PdsRepoLister, } impl DeepDiscoveryProducer { async fn run(mut self) -> Result<()> { loop { - self.enabled.wait_enabled("relay-first deep crawler").await; + self.lister + .enabled + .wait_enabled(self.lister.component) + .await; let pass = match self.wait_for_unhandled_pass().await { Ok(pass) => pass, Err(err) => { @@ -129,8 +129,11 @@ impl DeepDiscoveryProducer { let pass_key = keys::relay_first_pass_key(self.relay_url.as_str()); let handled_key = keys::relay_first_handled_key(self.relay_url.as_str()); loop { - self.enabled.wait_enabled("relay-first deep crawler").await; - let state = self.checker.state.clone(); + self.lister + .enabled + .wait_enabled(self.lister.component) + .await; + let state = self.lister.checker.state.clone(); let pass_key = pass_key.clone(); let handled_key = handled_key.clone(); let (pass, handled) = state @@ -150,7 +153,7 @@ impl DeepDiscoveryProducer { tokio::select! { _ = tokio::time::sleep(PASS_POLL_INTERVAL) => {} - changed = self.enabled.changed() => { + changed = self.lister.enabled.changed() => { if changed.is_err() { miette::bail!("crawler enable channel closed"); } @@ -172,7 +175,10 @@ impl DeepDiscoveryProducer { if page_count > MAX_LIST_HOSTS_PAGES { miette::bail!("listHosts exceeded {MAX_LIST_HOSTS_PAGES} pages"); } - self.enabled.wait_enabled("relay-first deep crawler").await; + self.lister + .enabled + .wait_enabled(self.lister.component) + .await; let url = pds_discovery::list_hosts_url(&base_url(&self.relay_url)?, cursor.as_deref()); let response = trusted_http() .get(url.clone()) @@ -196,7 +202,7 @@ impl DeepDiscoveryProducer { { log_pds_crawl_result(result); } - let producer = self.clone_for_host(); + let producer = self.clone(); crawls.spawn(async move { let hostname = host.hostname.to_string(); (hostname, producer.crawl_host(pass, host).await) @@ -230,7 +236,7 @@ impl DeepDiscoveryProducer { }, ] }; - self.send_cursor_updates(cursor_updates).await?; + self.lister.send_cursor_updates(cursor_updates).await?; let Some(next) = next_cursor else { return Ok(()); @@ -239,17 +245,6 @@ impl DeepDiscoveryProducer { } } - fn clone_for_host(&self) -> Self { - Self { - relay_url: self.relay_url.clone(), - checker: self.checker.clone(), - in_flight: self.in_flight.clone(), - tx: self.tx.clone(), - enabled: self.enabled.clone(), - stats: self.stats.clone(), - } - } - async fn crawl_host(mut self, pass: u64, host: Host) -> Result<()> { let hostname = host.hostname.as_str(); if matches!( @@ -269,7 +264,14 @@ impl DeepDiscoveryProducer { let url_host = public_host.as_str(); let pds = public_host.https_endpoint().url().clone(); let http = public_http(); - if self.checker.state.pds_meta.load().is_banned(url_host) { + if self + .lister + .checker + .state + .pds_meta + .load() + .is_banned(url_host) + { debug!(%hostname, "locally banned PDS, skipping deep discovery"); return Ok(()); } @@ -297,163 +299,26 @@ impl DeepDiscoveryProducer { trace!(%hostname, pass, "deep PDS crawl already complete"); return Ok(()); } - let mut cursor = saved.and_then(|state| state.cursor); - let mut page_count = 0usize; - - loop { - page_count += 1; - if page_count > MAX_PDS_PAGES { - miette::bail!("PDS listRepos exceeded {MAX_PDS_PAGES} pages"); - } - self.enabled.wait_enabled("relay-first deep crawler").await; - if self.checker.state.pds_meta.load().is_banned(hostname) { - debug!(%hostname, "locally banned PDS, stopping deep discovery"); - return Ok(()); - } - let mut url = pds - .join("/xrpc/com.atproto.sync.listRepos") - .into_diagnostic()?; - url.query_pairs_mut().append_pair("limit", "1000"); - if let Some(value) = &cursor { - url.query_pairs_mut().append_pair("cursor", value); - } - - let Some(bytes) = self - .fetch_pds_page(pds, hostname, account_count, url, http) - .await? - else { - self.send_cursor_updates(vec![CursorUpdate { - key: cursor_key, - value: Some(serialize_cursor(&PassCursor::complete(pass))?), + let start = saved.and_then(|state| state.cursor); + self.lister + .list(pds, account_count, http, start, |next| { + let state = next.map_or_else( + || PassCursor::complete(pass), + |next| PassCursor::active(pass, Some(next.clone())), + ); + Ok(vec![CursorUpdate { + key: cursor_key.clone(), + value: Some(serialize_cursor(&state)?), }]) - .await?; - return Ok(()); - }; - let page = self.parse_deep_page(bytes).await?; - self.stats.record_crawled(page.count); - - let in_flight = self.in_flight.acquire(page.unknown_dids).await; - let filter = self.checker.state.filter.load_full(); - let confirmed = if filter.check_signals() && !in_flight.is_empty() { - let mut retry_batch = self.checker.state.db.inner.batch(); - let confirmed = self - .checker - .check_signals_batch(in_flight, &filter, &mut retry_batch, &HashMap::new()) - .await?; - self.checker - .state - .db - .run(move |_db| retry_batch.commit().into_diagnostic()) - .await?; - confirmed - } else { - in_flight - }; - - let next_cursor = page.cursor; - if next_cursor.is_some() && next_cursor == cursor { - miette::bail!("PDS listRepos returned the same cursor twice"); - } - let persisted = next_cursor.as_ref().map_or_else( - || PassCursor::complete(pass), - |next| PassCursor::active(pass, Some(next.clone())), - ); - self.tx - .send(CrawlerBatch { - guards: confirmed, - listings: Vec::new(), - cursor_updates: vec![CursorUpdate { - key: cursor_key.clone(), - value: Some(serialize_cursor(&persisted)?), - }], - }) - .await - .into_diagnostic() - .wrap_err("crawler worker exited")?; - - let Some(next) = next_cursor else { - return Ok(()); - }; - cursor = Some(next); - } - } - - async fn fetch_pds_page( - &self, - pds: &Url, - hostname: &str, - account_count: u64, - url: Url, - http: &PublicHttpClient, - ) -> Result> { - let throttle = self.checker.throttler.get_handle(pds).await; - if throttle.is_throttled() { - miette::bail!("PDS is throttled until {}", throttle.throttled_until()); - } - let _permit = throttle.acquire().await; - let tier = self.checker.state.resolve_pds_tier(hostname); - throttle.wait_for_allow(account_count, &tier).await; - if throttle.is_throttled() { - miette::bail!("PDS became throttled while waiting for pacing"); - } - - let endpoint = PublicEndpoint::parse_http(&url) - .into_diagnostic() - .wrap_err("unsafe PDS endpoint in deep discovery")?; - let response = match http.get(&endpoint).timeout(throttle.timeout()).send().await { - Ok(response) => response, - Err(err) => { - throttle.record_failure_detail("deep_list_repos", err.to_string()); - return Err(err).into_diagnostic(); - } - }; - if response.status() == StatusCode::TOO_MANY_REQUESTS { - throttle.record_ratelimit(parse_retry_after(&response)); - miette::bail!("PDS rate limited deep discovery"); - } - if matches!( - response.status(), - StatusCode::BAD_REQUEST | StatusCode::NOT_FOUND | StatusCode::METHOD_NOT_ALLOWED - ) || response.status().is_redirection() - { - debug!(status = %response.status(), "PDS does not support listRepos; skipping"); - throttle.record_success(); - return Ok(None); - } - let response = match response.error_for_status() { - Ok(response) => response, - Err(err) => { - throttle.record_failure_detail("deep_list_repos", err.to_string()); - return Err(err).into_diagnostic(); - } - }; - let bytes = crate::util::read_limited_bytes(response, MAX_LIST_RESPONSE_BYTES).await?; - throttle.record_success(); - Ok(Some(bytes)) - } - - async fn parse_deep_page(&self, bytes: bytes::Bytes) -> Result { - let repos = self.checker.state.db.repos.clone(); - let filter = self.checker.state.db.filter.clone(); - let crawler = self.checker.state.db.crawler.clone(); - tokio::time::timeout( - BLOCKING_TASK_TIMEOUT, - tokio::task::spawn_blocking(move || { - let output = serde_json::from_slice::(&bytes) - .into_diagnostic() - .wrap_err("failed to parse PDS listRepos response")? - .into_static(); - select_unknown_repos(&repos, &filter, &crawler, output) - }), - ) - .await - .map_err(|_| miette::miette!("PDS listRepos parse timed out"))? - .into_diagnostic()? + }) + .await?; + Ok(()) } async fn load_cursor(&self, key: &[u8], pass: u64) -> Result> { let key = key.to_vec(); - self.checker + self.lister + .checker .state .db .run(move |db| { @@ -466,18 +331,6 @@ impl DeepDiscoveryProducer { }) .await } - - async fn send_cursor_updates(&self, cursor_updates: Vec) -> Result<()> { - self.tx - .send(CrawlerBatch { - guards: Vec::new(), - listings: Vec::new(), - cursor_updates, - }) - .await - .into_diagnostic() - .wrap_err("crawler worker exited") - } } fn log_pds_crawl_result(result: std::result::Result<(String, Result<()>), tokio::task::JoinError>) { @@ -490,39 +343,6 @@ fn log_pds_crawl_result(result: std::result::Result<(String, Result<()>), tokio: } } -struct DeepPage { - unknown_dids: Vec, - cursor: Option, - count: usize, -} - -fn select_unknown_repos( - repos: &fjall::Keyspace, - filter: &fjall::Keyspace, - crawler: &fjall::Keyspace, - output: ListReposOutput, -) -> Result { - let count = output.repos.len(); - let cursor = output.cursor; - let mut unknown_dids = Vec::new(); - for repo in output.repos { - let exclude_key = crate::db::filter::exclude_key(repo.did.as_str())?; - let did_key = keys::repo_key(&repo.did); - let retry_key = keys::crawler_retry_key(&repo.did); - let known = filter.contains_key(exclude_key).into_diagnostic()? - || repos.contains_key(did_key).into_diagnostic()? - || crawler.contains_key(retry_key).into_diagnostic()?; - if !known { - unknown_dids.push(repo.did); - } - } - Ok(DeepPage { - unknown_dids, - cursor, - count, - }) -} - fn serialize_cursor(cursor: &PassCursor) -> Result> { rmp_serde::to_vec(cursor).into_diagnostic() } @@ -553,11 +373,6 @@ mod tests { const HEAD: &str = "bafyreidf747c4x3lps3k4n357l3a3r57k3k465743k573k465743k5"; - fn list_repos_output(value: serde_json::Value) -> Result { - let bytes = serde_json::to_vec(&value).into_diagnostic()?; - serde_json::from_slice::(&bytes).into_diagnostic() - } - #[test] fn pass_cursor_round_trips() -> Result<()> { let cursor = PassCursor::active(7, Some("stinkpot".into())); @@ -577,67 +392,6 @@ mod tests { Ok(()) } - #[test] - fn deep_pages_select_only_unknown_repos_without_trusting_status() -> Result<()> { - let tmp = tempdir().into_diagnostic()?; - let config = Config { - database_path: tmp.path().to_path_buf(), - ..Default::default() - }; - let state = AppState::new(&config)?; - let known: Did = Did::new_static("did:web:known.example")?; - let excluded: Did = Did::new_static("did:web:excluded.example")?; - let retry: Did = Did::new_static("did:web:retry.example")?; - state - .db - .repos - .insert(keys::repo_key(&known), b"relay") - .into_diagnostic()?; - state - .db - .filter - .insert(crate::db::filter::exclude_key(excluded.as_str())?, b"") - .into_diagnostic()?; - state - .db - .crawler - .insert(keys::crawler_retry_key(&retry), b"") - .into_diagnostic()?; - - let page = select_unknown_repos( - &state.db.repos, - &state.db.filter, - &state.db.crawler, - list_repos_output(json!({ - "cursor": "next", - "repos": [ - {"did": known, "head": HEAD, "rev": "3j6s746574657", "active": false, "status": "takendown"}, - {"did": excluded, "head": HEAD, "rev": "3j6s746574657"}, - {"did": retry, "head": HEAD, "rev": "3j6s746574657"}, - {"did": "did:web:inactive-unknown.example", "head": HEAD, "rev": "3j6s746574657", "active": false, "status": "deactivated"} - ] - }))?, - )?; - - assert_eq!(page.count, 4); - assert_eq!(page.cursor.as_deref(), Some("next")); - assert_eq!(page.unknown_dids.len(), 1); - assert_eq!( - page.unknown_dids[0].as_str(), - "did:web:inactive-unknown.example" - ); - assert_eq!( - state - .db - .repos - .get(keys::repo_key(&known)) - .into_diagnostic()? - .as_deref(), - Some(b"relay".as_slice()) - ); - Ok(()) - } - #[tokio::test] async fn pds_pages_resume_and_commit_only_after_terminal_page() -> Result<()> { let requests = Arc::new(AtomicUsize::new(0)); @@ -697,15 +451,18 @@ mod tests { let pds = Url::parse("http://pds.example/").into_diagnostic()?; let mut producer = DeepDiscoveryProducer { relay_url: Url::parse("https://relay.example").into_diagnostic()?, - checker: SignalChecker { - http: PublicHttpClient::test(reqwest::Client::new()), - state: state.clone(), - throttler: state.throttler.clone(), + lister: PdsRepoLister { + checker: SignalChecker { + http: PublicHttpClient::test(reqwest::Client::new()), + state: state.clone(), + throttler: state.throttler.clone(), + }, + in_flight: InFlight::new(), + tx, + enabled: state.crawler_enabled.subscribe(), + stats, + component: "relay-first deep crawler", }, - in_flight: InFlight::new(), - tx, - enabled: state.crawler_enabled.subscribe(), - stats, }; let http = PublicHttpClient::test( @@ -812,15 +569,18 @@ mod tests { let (tx, _rx) = mpsc::channel(1); let mut producer = DeepDiscoveryProducer { relay_url: Url::parse("https://relay.example").into_diagnostic()?, - checker: SignalChecker { - http: PublicHttpClient::test(reqwest::Client::new()), - state: state.clone(), - throttler: state.throttler.clone(), + lister: PdsRepoLister { + checker: SignalChecker { + http: PublicHttpClient::test(reqwest::Client::new()), + state: state.clone(), + throttler: state.throttler.clone(), + }, + in_flight: InFlight::new(), + tx, + enabled: state.crawler_enabled.subscribe(), + stats, + component: "relay-first deep crawler", }, - in_flight: InFlight::new(), - tx, - enabled: state.crawler_enabled.subscribe(), - stats, }; assert!(