[crawler] resync known repos a newly followed pds lists at a newer rev
110 unmodified linesExpand all11111211311411511511611711811956 unmodified lines1761771781791801801811821831843 unmodified lines1881891901911921931941941951961971981992002012022022032042052062032042052062072082092102115 unmodified lines21721821922022122222322422560 unmodified lines28628728828929029129229329429529629729829930030130230330430530630730830931031131231331431531631731831932032132243 unmodified lines36636736836937037137237337437537637737837938038138238338438538638738838939039139239339439539639739839940040140240340440540640740840941031 unmodified lines110 unmodified linesExpand all let http = self.http.clone(); // we don't know its size, and one account paces it at the tier's base rate let listing = self .lister .list(pds, 1, &http, cursor, |next| { .list(pds, 1, &http, cursor, true, |next| { Ok(vec![CursorUpdate { key: key.clone(), // the last page deletes the key in the same commit as its repos value: next56 unmodified lines use crate::state::AppState; use axum::extract::{Query, State}; use axum::response::{IntoResponse, Response}; use axum::{Json, Router, http::StatusCode, routing::get}; use jacquard_common::types::string::Did; use jacquard_common::types::string::{Did, Tid}; use serde_json::json; use std::collections::HashMap; use std::sync::Arc; use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering};3 unmodified lines
struct MockPds { requests: AtomicUsize, fail_second_page: AtomicBool, known: std::sync::Mutex<Vec<serde_json::Value>>, }
/// serves two `listRepos` pages, one unknown repo on each. /// serves two `listRepos` pages, one unknown repo on each, with `known` on the first. async fn mock_pds() -> Result<(std::net::SocketAddr, Arc<MockPds>)> { async fn list_repos( State(pds): State<Arc<MockPds>>, Query(query): Query<HashMap<String, String>>, ) -> Response { pds.requests.fetch_add(1, Ordering::SeqCst); if !query.contains_key("cursor") { return Json(json!({ "cursor": "next", "repos": [{"did": "did:web:first.example", "head": HEAD, "rev": "3j6s746574657"}] })) .into_response(); let mut repos = vec![ json!({"did": "did:web:first.example", "head": HEAD, "rev": "3j6s746574657"}), ]; repos.extend(pds.known.lock().unwrap().iter().cloned()); return Json(json!({ "cursor": "next", "repos": repos })).into_response(); } if pds.fail_second_page.load(Ordering::SeqCst) { return StatusCode::INTERNAL_SERVER_ERROR.into_response(); }5 unmodified lines
let pds = Arc::new(MockPds { requests: AtomicUsize::new(0), fail_second_page: AtomicBool::new(false), known: std::sync::Mutex::new(Vec::new()), }); let app = Router::new() .route("/xrpc/com.atproto.sync.listRepos", get(list_repos)) .with_state(pds.clone());60 unmodified lines .contains_key(keys::repo_key(&Did::new_static(did)?)) .into_diagnostic() }
/// stores `did` as synced at `rev`, its doc pointing at `pds`. fn synced_at(state: &AppState, did: &'static str, rev: &str, pds: &'static str) -> Result<()> { let mut repo = crate::types::RepoState::synced(); repo.root = Some(crate::types::Commit { version: 3, rev: crate::db::types::DbTid::from(&Tid::new(rev).into_diagnostic()?), data: crate::car::compute_cid(b"root"), prev: None, sig: bytes::Bytes::new(), }); repo.pds = Some(jacquard_common::CowStr::Borrowed(pds)); let mut txn = crate::db::Txn::new(&state.db); txn.put_repo(&Did::new_static(did)?, &repo)?; txn.commit() }
fn backfill_queued(state: &AppState, did: &'static str) -> Result<bool> { let key = keys::repo_metadata_key(&Did::new_static(did)?); let Some(bytes) = state.db.repo_metadata.get(key).into_diagnostic()? else { return Ok(false); }; let metadata = crate::db::deser_repo_meta(&bytes)?; state .db .indexer .is_pending(&keys::pending_key(metadata.index_id)) .into_diagnostic() }
/// the worker commits batches on its own task, so wait for it to catch up. async fn eventually(check: impl Fn() -> Result<bool>) -> Result<()> { tokio::time::timeout(Duration::from_secs(5), async { while !check()? {43 unmodified lines Ok(()) }
#[tokio::test] async fn known_repos_listed_at_a_newer_rev_get_resynced() -> Result<()> { let dir = tempdir().into_diagnostic()?; let state = test_state(&dir)?; let (addr, pds) = mock_pds().await?; let crawler = crawler(&state, addr)?; let (old, new) = ("3mz2ipdzmrm22", "3mz2j7sg2kh22"); synced_at(&state, "did:web:moved.example", old, "https://pds.example/")?; synced_at( &state, "did:web:current.example", new, "https://pds.example/", )?; synced_at( &state, "did:web:elsewhere.example", old, "https://other.example/", )?; *pds.known.lock().unwrap() = vec![ json!({"did": "did:web:moved.example", "head": HEAD, "rev": new}), json!({"did": "did:web:current.example", "head": HEAD, "rev": new}), json!({"did": "did:web:elsewhere.example", "head": HEAD, "rev": new}), ]; queue(&state, "pds.example", None)?;
let url = Url::parse("http://pds.example/").into_diagnostic()?; let key = keys::new_pds_crawl_key("pds.example"); crawler.list_host("pds.example", key, &url, None).await?; eventually(|| Ok(load_queued(&state.db)?.is_empty())).await?; assert!(backfill_queued(&state, "did:web:moved.example")?); assert!(!backfill_queued(&state, "did:web:current.example")?); assert!(!backfill_queued(&state, "did:web:elsewhere.example")?); Ok(()) }
#[tokio::test] async fn banned_unfollowed_and_private_hosts_drop_their_crawl() -> Result<()> { let dir = tempdir().into_diagnostic()?; let state = test_state(&dir)?; let (addr, pds) = mock_pds().await?;31 unmodified lines12344567891011121313141516171819202122 unmodified lines4445464748495051525354555657585925 unmodified lines85868788858990919293949535 unmodified lines13113213313412913513613713813960 unmodified lines2002012022031982042052062072081 unmodified line21021121221320821421521621721845 unmodified lines2642652662672682692702712721 unmodified line27427527627727827928028128228328428527728628728828929029129229329429527928028128229629729829930030130230330430530630730830931031131231331431531631731831932032132232332432532632732832933033159 unmodified lines39139239339439539639739839961 unmodified linesuse std::collections::HashMap;use std::time::Duration;
use super::list_repos::producer::ListReposOutput;use super::list_repos::producer::{ListReposOutput, ListedRepo};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::worker::{CrawlerBatch, CursorUpdate, RepoListing};use super::{CrawlerStats, InFlight, SignalChecker};use crate::db::counts::pds_host;use crate::db::keys;use crate::db::types::DbTid;use crate::net::{PublicEndpoint, PublicHttpClient};use crate::state::AppState;use crate::util::{WatchEnabledExt, parse_retry_after};
22 unmodified lines
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. /// `resync_newer` also resyncs the known repos of this host it lists at a newer rev. pub(crate) async fn list( &mut self, pds: &Url, account_count: u64, http: &PublicHttpClient, mut cursor: Option<SmolStr>, resync_newer: bool, progress: impl Fn(Option<&SmolStr>) -> Result<Vec<CursorUpdate>>, ) -> Result<PdsListing> { let hostname = pds .host_str()25 unmodified lines else { self.send_cursor_updates(progress(None)?).await?; return Ok(PdsListing::Complete); }; let page = self.parse_page(bytes).await?; let page = self .parse_page(bytes, resync_newer.then(|| SmolStr::new(hostname))) .await?; self.stats.record_crawled(page.count);
let state = &self.checker.state; let mut unknown = page.unknown_dids;35 unmodified lines admitted += confirmed.len() as u64; self.tx .send(CrawlerBatch { guards: confirmed, listings: Vec::new(), listings: page.newer, cursor_updates: progress(next_cursor.as_ref())?, applied: None, }) .await60 unmodified lines throttle.record_success(); Ok(Some(bytes)) }
async fn parse_page(&self, bytes: bytes::Bytes) -> Result<Page> { async fn parse_page(&self, bytes: bytes::Bytes, resync_host: Option<SmolStr>) -> Result<Page> { 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(1 unmodified line tokio::task::spawn_blocking(move || { let output = serde_json::from_slice::<ListReposOutput>(&bytes) .into_diagnostic() .wrap_err("failed to parse PDS listRepos response")?; select_unknown_repos(&repos, &filter, &crawler, output) select_unknown_repos(&repos, &filter, &crawler, output, resync_host.as_deref()) }), ) .await .map_err(|_| miette::miette!("PDS listRepos parse timed out"))?45 unmodified lines}
struct Page { unknown_dids: Vec<Did>, newer: Vec<RepoListing>, cursor: Option<SmolStr>, count: usize,}
1 unmodified line repos: &fjall::Keyspace, filter: &fjall::Keyspace, crawler: &fjall::Keyspace, output: ListReposOutput, resync_host: Option<&str>,) -> Result<Page> { let count = output.repos.len(); let cursor = output.cursor; let mut unknown_dids = Vec::new(); let mut newer = 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); if filter.contains_key(exclude_key).into_diagnostic()? { continue; } if let Some(stored) = repos.get(keys::repo_key(&repo.did)).into_diagnostic()? { if let Some(host) = resync_host { newer.extend(listing_if_newer(host, repo, &stored)?); } continue; } 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 { if !crawler.contains_key(retry_key).into_diagnostic()? { unknown_dids.push(repo.did); } } Ok(Page { unknown_dids, newer, cursor, count, })}
/// `repo` as a listing if its doc points at `host` and it's listed past the rev we holdfn listing_if_newer( host: &str, repo: ListedRepo<'_>, stored: &[u8],) -> Result<Option<RepoListing>> { let stored = crate::db::deser_repo_state(stored)?; if stored.pds.as_deref().and_then(pds_host).as_deref() != Some(host) { return Ok(None); } let listing = RepoListing { did: repo.did.into_static(), rev: DbTid::from(&repo.rev), active: repo.active.unwrap_or(true), status: repo.status.into_static(), create_if_missing: false, }; Ok(listing.claims_newer(&stored).then_some(listing))}
#[cfg(test)]mod tests { use super::*; use crate::config::Config;59 unmodified lines {"did": retry, "head": HEAD, "rev": "3j6s746574657"}, {"did": "did:web:inactive-unknown.example", "head": HEAD, "rev": "3j6s746574657", "active": false, "status": "deactivated"} ] }))?, None, )?;
assert_eq!(page.count, 4); assert_eq!(page.cursor.as_deref(), Some("next"));61 unmodified lines298 unmodified linesExpand all299300301302303303304305306307339 unmodified linesExpand all298 unmodified linesExpand all return Ok(()); } let start = saved.and_then(|state| state.cursor); self.lister .list(pds, account_count, http, start, |next| { .list(pds, account_count, http, start, false, |next| { let state = next.map_or_else( || PassCursor::complete(pass), |next| PassCursor::active(pass, Some(next.clone())), );339 unmodified linesExpand all