[crawler] cap a listRepos page at 512 KiB and the repos we asked for
400 unmodified linesExpand all401402403404405406407408409410411412413414415416417418419420421422423424425426427428429221 unmodified linesExpand all400 unmodified linesExpand all Ok(()) }
#[tokio::test] async fn a_page_over_the_size_cap_breaks_the_walk() -> Result<()> { let dir = tempdir().into_diagnostic()?; let state = test_state(&dir)?; let (addr, pds) = mock_pds().await?; let crawler = crawler(&state, addr)?; *pds.known.lock().unwrap() = vec![json!({ "did": "did:web:padded.example", "head": HEAD, "rev": "3j6s746574657", "padding": "x".repeat(512 * 1024), })]; 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("too large")); assert!(!tracked(&state, "did:web:first.example")?); 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 all20 unmodified lines21222324252526272829303155 unmodified lines8788899089919293949596225 unmodified linesExpand all322323324325326327328329330331332102 unmodified linesExpand all435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465150 unmodified linesExpand all20 unmodified linesuse crate::state::AppState;use crate::types::{RepoState, RepoStatus, ResyncState};use crate::util::{WatchEnabledExt, parse_retry_after};
/// an honest page of 1000 repos is about 145 KiBconst MAX_LIST_RESPONSE_BYTES: usize = 10 * 1024 * 1024;const MAX_LIST_RESPONSE_BYTES: usize = 512 * 1024;const LIST_LIMIT: usize = 1000;const BLOCKING_TASK_TIMEOUT: Duration = Duration::from_secs(30);const MAX_PAGES: usize = 100_000;
/// how a walk over one PDS ended.55 unmodified lines } let mut url = pds .join("/xrpc/com.atproto.sync.listRepos") .into_diagnostic()?; url.query_pairs_mut().append_pair("limit", "1000"); url.query_pairs_mut() .append_pair("limit", &LIST_LIMIT.to_string()); if let Some(value) = &cursor { url.query_pairs_mut().append_pair("cursor", value); }
225 unmodified linesExpand all host: &str, skew_secs: i64,) -> Result<Page> { let count = output.repos.len(); if count > LIST_LIMIT { miette::bail!("PDS listRepos returned {count} repos for a limit of {LIST_LIMIT}"); } let cursor = output.cursor; let mut unknown_dids = Vec::new(); let mut newer = Vec::new(); let mut stuck = Vec::new();102 unmodified linesExpand all }) }
#[test] fn a_page_with_more_repos_than_the_limit_is_refused() -> Result<()> { let tmp = tempdir().into_diagnostic()?; let state = AppState::new(&Config { database_path: tmp.path().to_path_buf(), ..Default::default() })?; for (count, fits) in [(LIST_LIMIT, true), (LIST_LIMIT + 1, false)] { let repos: Vec<_> = (0..count) .map(|i| json!({"did": format!("did:web:r{i}.example"), "head": HEAD, "rev": "3j6s746574657"})) .collect(); let page = select_unknown_repos( &state.db, list_repos_output(json!({ "repos": repos }))?, KnownRepos::Skip, "pds.example", state.rev_clock_skew_secs, ); assert_eq!(page.is_ok(), fits, "{count} repos"); } Ok(()) }
#[test] fn a_listed_rev_from_the_future_is_not_newer() -> Result<()> { let tmp = tempdir().into_diagnostic()?; let state = AppState::new(&Config { database_path: tmp.path().to_path_buf(),150 unmodified linesExpand all