[crawler] end a listRepos walk at any cursor it already saw
421 unmodified linesExpand all422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466221 unmodified linesExpand all421 unmodified linesExpand all Ok(()) }
#[tokio::test] async fn a_walk_ends_when_a_cursor_comes_round_again() -> Result<()> { async fn list_repos( State(requests): State<Arc<AtomicUsize>>, Query(query): Query<HashMap<String, String>>, ) -> Json<serde_json::Value> { requests.fetch_add(1, Ordering::SeqCst); let next = match query.get("cursor").map(String::as_str) { None => "a", Some("a") => "b", _ => "a", }; Json(json!({ "cursor": next, "repos": [] })) }
let dir = tempdir().into_diagnostic()?; let state = test_state(&dir)?; let requests = Arc::new(AtomicUsize::new(0)); let app = Router::new() .route("/xrpc/com.atproto.sync.listRepos", get(list_repos)) .with_state(requests.clone()); let listener = tokio::net::TcpListener::bind("127.0.0.1:0") .await .into_diagnostic()?; let addr = listener.local_addr().into_diagnostic()?; tokio::spawn(async move { axum::serve(listener, app).await }); let crawler = crawler(&state, addr)?;
let url = Url::parse("http://pds.example/").into_diagnostic()?; let result = crawler .list_host("pds.example", Walk::New, &url, None) .await; assert!(format!("{:?}", result.unwrap_err()).contains("already returned")); assert_eq!(requests.load(Ordering::SeqCst), 3); 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?;221 unmodified linesExpand all11234567 unmodified lines73747576777879808160 unmodified lines142143144145145146146147148149150151152153465 unmodified linesExpand alluse std::collections::HashMap;use std::collections::{HashMap, HashSet};use std::time::Duration;
use super::list_repos::producer::{ListReposOutput, ListedRepo};use jacquard_common::{IntoStatic, types::string::Did};67 unmodified lines .host_str() .ok_or_else(|| miette::miette!("PDS URL has no hostname"))?; let mut page_count = 0usize; let mut admitted = 0u64; let mut seen: HashSet<u64> = cursor.iter().map(crate::util::hash).collect();
loop { page_count += 1; if page_count > MAX_PAGES {60 unmodified lines debug!(%hostname, skipped, "account limit reached for host, skipping crawled accounts"); self.stats.record_over_limit(skipped); } let next_cursor = page.cursor; if next_cursor.is_some() && next_cursor == cursor { miette::bail!("PDS listRepos returned the same cursor twice"); if let Some(next) = &next_cursor && !seen.insert(crate::util::hash(next)) { miette::bail!("PDS listRepos returned a cursor it already returned"); } admitted += confirmed.len() as u64; self.tx .send(CrawlerBatch {465 unmodified linesExpand all