[crawler] rewalk a followed pds that asks for a crawl and resync its stuck repos
78 unmodified lines798081828384858687888990134 unmodified linesExpand all225226227228229230231232233234235236399 unmodified linesExpand all78 unmodified lines #[cfg(feature = "indexer")] pub backfill_notify: Notify, #[cfg(feature = "indexer")] pub(crate) new_pds_crawl_notify: Notify, /// how long a host's rewalk allowance lasts, see [`crate::control::new_pds_crawl`] #[cfg(feature = "indexer")] pub(crate) rewalk_window: Duration, #[cfg(feature = "indexer")] pub crawler_enabled: watch::Sender<bool>, #[cfg(feature = "indexer")] pub backfill_enabled: watch::Sender<bool>, #[cfg(feature = "indexer")]134 unmodified linesExpand all backfill_notify: Notify::new(), #[cfg(feature = "indexer")] new_pds_crawl_notify: Notify::new(), #[cfg(feature = "indexer")] rewalk_window: config .offline_host_retry_interval .unwrap_or(crate::control::new_pds_crawl::DEFAULT_REWALK_WINDOW), #[cfg(feature = "indexer")] crawler_enabled, firehose_enabled, #[cfg(feature = "indexer")] backfill_enabled,399 unmodified linesExpand all439 unmodified linesExpand all4404414424434444454464474484494504514 unmodified lines456457458459460461462463464465466467468469470471472473474475476477478479480413 unmodified linesExpand all89489589689789889990090190290390490590690790890991091191291391491591691791891992092192292392492592692792892993093193293393493593693793893994094194294394494594694794894995095195295395495595695795895996096196296396496596696796896997097197297397497597697797897998098198298398498598698798898999099199299399499599699799899910001001100210031004100510061007100810091010101110121013101410151016101710181019102010211022102310241025102610271028102910301031103210331034103510361037103810391040104110421043104410451046104710481049105010511052506 unmodified linesExpand all439 unmodified linesExpand all if let Some(early) = self .check_admission_preconditions(&url, host.as_str(), true) .await? { #[cfg(feature = "indexer")] if early.is_ok() { self.request_rewalk(host.as_str()).await?; } return Ok(early); }
let Some(reservation) = self.state.pds_daily_limit.reserve() else {4 unmodified lines self.spawn_admitted_source(&url, host.as_str(), false, reservation, true) .await }
#[cfg(feature = "indexer")] async fn request_rewalk(&self, host: &str) -> Result<()> { let now = chrono::Utc::now().timestamp(); let window = self.state.rewalk_window.as_secs() as i64; let walked = host.to_owned(); let rewalk = self .state .db .run(move |db| new_pds_crawl::request_rewalk(db, &walked, now, window)) .await?; debug!(%host, ?rewalk, "requested a rewalk"); if rewalk == new_pds_crawl::Rewalk::Queued { new_pds_crawl::wake(&self.state); } Ok(()) }
async fn admit_seeded_source(&self, url: Url, queue_crawl: bool) -> Result<PdsAdmissionResult> { let (host, url) = match canonical_pds_url(&url) { Ok(source) => source, Err(error) => return Ok(Err(PdsAdmissionRejection::InvalidHost(error))),413 unmodified linesExpand all assert!(!queued("fresh.example")?); Ok(()) }
#[cfg(feature = "indexer")] fn walking(hydrant: &Hydrant, host: &str) -> Result<bool> { let cursors = &hydrant.state.db.cursors; Ok(cursors .contains_key(keys::new_pds_crawl_key(host)) .into_diagnostic()? || cursors .contains_key(keys::rewalk_key(host)) .into_diagnostic()?) }
#[cfg(feature = "indexer")] fn finish_walk(hydrant: &Hydrant, host: &str) -> Result<()> { let cursors = &hydrant.state.db.cursors; cursors .remove(keys::new_pds_crawl_key(host)) .into_diagnostic()?; cursors.remove(keys::rewalk_key(host)).into_diagnostic() }
#[cfg(feature = "indexer")] #[tokio::test] async fn a_known_host_gets_three_rewalks_a_window_outside_the_daily_limit() -> Result<()> { let dir = tempfile::tempdir().into_diagnostic()?; let hydrant = Hydrant::new(test_config_with_limit(&dir, 2)).await?; startable(&hydrant); hydrant.firehose.disable_spawns_for_test(); let url = Url::parse("wss://known.example/").into_diagnostic()?; assert_eq!( hydrant.firehose.add_source_validated(url.clone()).await?, Ok(PdsAdmission::StartedNew) ); finish_walk(&hydrant, "known.example")?;
for _ in 0..3 { assert_eq!( hydrant.firehose.add_source_validated(url.clone()).await?, Ok(PdsAdmission::RestartedKnown) ); assert!( hydrant .state .db .cursors .contains_key(keys::rewalk_key("known.example")) .into_diagnostic()? ); finish_walk(&hydrant, "known.example")?; } assert_eq!( hydrant.firehose.add_source_validated(url).await?, Ok(PdsAdmission::RestartedKnown) ); assert!(!walking(&hydrant, "known.example")?);
assert!(matches!( crate::db::load_pds_daily_adds(&hydrant.state.db)?, Some((_, 1)) )); let other = Url::parse("wss://other.example/").into_diagnostic()?; assert_eq!( hydrant.firehose.add_source_validated(other).await?, Ok(PdsAdmission::StartedNew) ); Ok(()) }
#[cfg(feature = "indexer")] #[tokio::test] async fn requests_during_a_walk_neither_queue_nor_count() -> Result<()> { let dir = tempfile::tempdir().into_diagnostic()?; let hydrant = Hydrant::new(test_config(&dir)).await?; startable(&hydrant); hydrant.firehose.disable_spawns_for_test(); let url = Url::parse("wss://busy.example/").into_diagnostic()?; assert_eq!( hydrant.firehose.add_source_validated(url.clone()).await?, Ok(PdsAdmission::StartedNew) );
// the new host's own walk, then a rewalk, both still running for walk in [keys::new_pds_crawl_key, keys::rewalk_key] { hydrant .state .db .cursors .insert( walk("busy.example"), crate::control::new_pds_crawl::encode_cursor(Some(&"page".into()))?, ) .into_diagnostic()?; for _ in 0..5 { assert_eq!( hydrant.firehose.add_source_validated(url.clone()).await?, Ok(PdsAdmission::RestartedKnown) ); } assert_eq!( crate::control::new_pds_crawl::decode_cursor( &hydrant .state .db .cursors .get(walk("busy.example")) .into_diagnostic()? .expect("the running walk stays queued") )? .as_deref(), Some("page"), "a request must not restart a running walk" ); finish_walk(&hydrant, "busy.example")?; }
for _ in 0..3 { assert_eq!( hydrant.firehose.add_source_validated(url.clone()).await?, Ok(PdsAdmission::RestartedKnown) ); assert!( walking(&hydrant, "busy.example")?, "the allowance is still whole" ); finish_walk(&hydrant, "busy.example")?; } Ok(()) }
#[cfg(feature = "indexer")] #[tokio::test] async fn a_banned_known_host_gets_no_rewalk() -> Result<()> { let dir = tempfile::tempdir().into_diagnostic()?; let hydrant = Hydrant::new(test_config(&dir)).await?; startable(&hydrant); hydrant.firehose.disable_spawns_for_test(); let url = Url::parse("wss://turned.example/").into_diagnostic()?; assert_eq!( hydrant.firehose.add_source_validated(url.clone()).await?, Ok(PdsAdmission::StartedNew) ); finish_walk(&hydrant, "turned.example")?;
hydrant.pds.ban("turned.example").await?; assert_eq!( hydrant.firehose.add_source_validated(url).await?, Err(PdsAdmissionRejection::HostBanned) ); assert!(!walking(&hydrant, "turned.example")?); Ok(()) }
/// `upsert_async` returns the previous handle, whose drop cancels its task. #[tokio::test] async fn replacing_a_source_cancels_the_previous_ingestor() -> Result<()> { let dir = tempfile::tempdir().into_diagnostic()?;506 unmodified linesExpand all112345678910111213141516172 unmodified lines202122232425262728293031323334353637383940414243444546474849505152535455565758596061626364656667686970717273747576777879808182838456 unmodified lines141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235//! the one-off `listRepos` walk an indexer owes a PDS it starts following.//! the one-off `listRepos` walk an indexer owes a PDS it starts following, and the//! rewalk a followed PDS can ask for through `requestCrawl`.//!//! a new PDS is tailed from the relay's seq, so repos that already lived on it//! never reach us as events and nothing else lists them. only hosts that are new//! to the network get the walk: a seed relay's hosts count as bootstrap until its//! `listHosts` has been read to the end once, otherwise a fresh node would walk//! every PDS out there. a relay has nothing to backfill, so there it's a no-op.//!//! a rewalk lists the host again to pick up repos we missed and resync the ones of//! its repos stuck desynchronized or failing. a host gets [`REWALKS_PER_WINDOW`] of//! them per offline host retry interval, and asking while one is queued or running//! does nothing.
#[cfg(feature = "indexer")]mod enabled { use miette::{IntoDiagnostic, Result};2 unmodified lines
use crate::db::{Db, keys}; use crate::state::AppState;
pub(crate) const REWALKS_PER_WINDOW: u8 = 3; /// the rewalk window when `OFFLINE_HOST_RETRY_INTERVAL` is off, its default pub(crate) const DEFAULT_REWALK_WINDOW: std::time::Duration = std::time::Duration::from_secs(30 * 60);
#[derive(Debug, Clone, Copy, PartialEq, Eq)] pub(crate) enum Rewalk { Queued, AlreadyWalking, OutOfAllowance, }
#[derive(serde::Serialize, serde::Deserialize)] struct Allowance { since: i64, used: u8, }
/// queue a rewalk of `host` unless it's walking already or used up its allowance for the /// `window` seconds since its first rewalk. call [`wake`] once it's queued. pub(crate) fn request_rewalk(db: &Db, host: &str, now: i64, window: i64) -> Result<Rewalk> { let walk_key = keys::rewalk_key(host); if db.cursors.contains_key(&walk_key).into_diagnostic()? || db .cursors .contains_key(keys::new_pds_crawl_key(host)) .into_diagnostic()? { return Ok(Rewalk::AlreadyWalking); } let allowance_key = keys::rewalk_allowance_key(host); let mut allowance = db .cursors .get(&allowance_key) .into_diagnostic()? .map(|bytes| rmp_serde::from_slice::<Allowance>(&bytes).into_diagnostic()) .transpose()? .filter(|allowance| now < allowance.since.saturating_add(window)) .unwrap_or(Allowance { since: now, used: 0, }); if allowance.used >= REWALKS_PER_WINDOW { return Ok(Rewalk::OutOfAllowance); } allowance.used += 1; let mut batch = db.inner.batch(); batch.insert(&db.cursors, walk_key, encode_cursor(None)?); batch.insert( &db.cursors, allowance_key, rmp_serde::to_vec(&allowance).into_diagnostic()?, ); batch.commit().into_diagnostic()?; Ok(Rewalk::Queued) }
/// stage the walk in the batch that admits the host, so a restart can't lose it. pub(crate) fn queue(batch: &mut fjall::OwnedWriteBatch, db: &Db, host: &str) -> Result<()> { batch.insert( &db.cursors,56 unmodified lines}
#[cfg(not(feature = "indexer"))]pub(crate) use noop::*;
#[cfg(all(test, feature = "indexer"))]mod tests { use super::*; use crate::config::Config; use crate::db::keys; use crate::state::AppState; use miette::{IntoDiagnostic, Result};
const WINDOW: i64 = 1800;
fn finish_walk(state: &AppState, host: &str) -> Result<()> { state .db .cursors .remove(keys::rewalk_key(host)) .into_diagnostic() }
#[test] fn the_allowance_comes_back_once_its_window_is_over() -> Result<()> { let dir = tempfile::tempdir().into_diagnostic()?; let state = AppState::new(&Config { database_path: dir.path().to_path_buf(), ..Default::default() })?; for at in [0, 10, 20] { assert_eq!( request_rewalk(&state.db, "pds.example", at, WINDOW)?, Rewalk::Queued ); finish_walk(&state, "pds.example")?; } assert_eq!( request_rewalk(&state.db, "pds.example", WINDOW - 1, WINDOW)?, Rewalk::OutOfAllowance ); assert_eq!( request_rewalk(&state.db, "other.example", WINDOW - 1, WINDOW)?, Rewalk::Queued, "another host's allowance is its own" ); assert_eq!( request_rewalk(&state.db, "pds.example", WINDOW, WINDOW)?, Rewalk::Queued ); Ok(()) }
#[test] fn the_allowance_and_a_queued_rewalk_survive_a_restart() -> Result<()> { let dir = tempfile::tempdir().into_diagnostic()?; let config = Config { database_path: dir.path().to_path_buf(), ..Default::default() }; let state = AppState::new(&config)?; for _ in 0..2 { assert_eq!( request_rewalk(&state.db, "pds.example", 0, WINDOW)?, Rewalk::Queued ); finish_walk(&state, "pds.example")?; } assert_eq!( request_rewalk(&state.db, "pds.example", 0, WINDOW)?, Rewalk::Queued ); state.db.persist()?; drop(state);
let state = AppState::new(&config)?; assert_eq!( request_rewalk(&state.db, "pds.example", 1, WINDOW)?, Rewalk::AlreadyWalking ); assert!( state .db .cursors .contains_key(keys::rewalk_key("pds.example")) .into_diagnostic()? ); finish_walk(&state, "pds.example")?; assert_eq!( request_rewalk(&state.db, "pds.example", 1, WINDOW)?, Rewalk::OutOfAllowance ); Ok(()) }}4 unmodified lines567899101112131 unmodified line151617181920192021222324252627282930313233343536373839404142434445464748495051525354555629 unmodified lines868788895890919293949596976698991001011021 unmodified line104105106107767710810911011111211313 unmodified lines127128129130991311321011331341351361371381071391401411421431441451461471151481491501511522 unmodified lines1551561571581261591601611621637 unmodified lines171172173174142143144145146147148149150175176177178179180181182183184185186187188189179 unmodified linesExpand all36937037137233837337437537637737834537938038138238338435138538638738838939039139235835939339439539639739829 unmodified lines4284294304313973984324334344354364374384394404414424434444454464474484494504514524534544554564574584594604614624634644654664674684694704714724734744754764774784794804814824834844854864874884894904914924934944954964974984995005015025035045055065075085095105115125135145155165175185195205215225235245255265275285295305315325335345355365375385395405415425435445455465475485495505515525535545555565575585595605615625635645655665675685695705715725735749 unmodified lines5845855865874244254264274284295885895905914335925935945955964355975985996006012 unmodified lines4 unmodified linesuse tokio::task::JoinSet;use tracing::{error, info, warn};use url::Url;
use super::pds_repos::{PdsListing, PdsRepoLister};use super::pds_repos::{KnownRepos, PdsListing, PdsRepoLister};use super::worker::CursorUpdate;use crate::control::new_pds_crawl;use crate::db::keys;use crate::net::{PublicHost, PublicHttpClient};1 unmodified line
const MAX_CONCURRENT_CRAWLS: usize = 4;const RETRY_DELAY: Duration = Duration::from_secs(5 * 60);
/// runs the one-off `listRepos` walk queued for each PDS we start following,/// see [`new_pds_crawl`]./// which walk a host is queued for, see [`new_pds_crawl`].#[derive(Debug, Clone, Copy, PartialEq, Eq)]enum Walk { New, Rewalk,}
impl Walk { const ALL: [Self; 2] = [Self::New, Self::Rewalk];
fn prefix(self) -> &'static [u8] { match self { Self::New => keys::NEW_PDS_CRAWL_PREFIX, Self::Rewalk => keys::REWALK_PREFIX, } }
fn key(self, host: &str) -> Vec<u8> { match self { Self::New => keys::new_pds_crawl_key(host), Self::Rewalk => keys::rewalk_key(host), } }
fn known_repos(self) -> KnownRepos { match self { Self::New => KnownRepos::ResyncNewer, Self::Rewalk => KnownRepos::ResyncStuck, } }}
/// runs the `listRepos` walks queued for the PDSes we start following and the/// rewalks followed ones asked for, see [`new_pds_crawl`].#[derive(Clone)]pub(crate) struct NewPdsCrawler { pub(crate) lister: PdsRepoLister, pub(crate) http: PublicHttpClient,29 unmodified lines async fn crawl_queued(&self) -> Result<bool> { let queued = self.lister.checker.state.db.run(load_queued).await?; let mut all_done = true; let mut crawls = JoinSet::new(); for (host, cursor) in queued { for (walk, host, cursor) in queued { if crawls.len() >= MAX_CONCURRENT_CRAWLS && let Some(result) = crawls.join_next().await { all_done &= crawl_succeeded(result); } let crawler = self.clone(); crawls.spawn(async move { let result = crawler.crawl(&host, cursor).await; let result = crawler.crawl(walk, &host, cursor).await; (host, result) }); } while let Some(result) = crawls.join_next().await {1 unmodified line } Ok(all_done) }
async fn crawl(self, host: &str, cursor: Option<SmolStr>) -> Result<()> { let key = keys::new_pds_crawl_key(host); async fn crawl(self, walk: Walk, host: &str, cursor: Option<SmolStr>) -> Result<()> { let key = walk.key(host); let public = match PublicHost::parse(host) { Ok(public) => public, Err(error) => { warn!(%host, %error, "queued PDS is not a public host, dropping its crawl");13 unmodified lines info!(%host, "PDS is no longer followed, dropping its crawl"); return self.forget(key).await; }
info!(%host, resuming = cursor.is_some(), "listing repos of a newly followed PDS"); info!(%host, ?walk, resuming = cursor.is_some(), "listing repos of a PDS"); let pds = public.https_endpoint().url().clone(); self.list_host(host, key, &pds, cursor).await self.list_host(host, walk, &pds, cursor).await }
async fn list_host( mut self, host: &str, key: Vec<u8>, walk: Walk, pds: &Url, cursor: Option<SmolStr>, ) -> Result<()> { let key = walk.key(host); 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, true, |next| { .list(pds, 1, &http, cursor, walk.known_repos(), |next| { Ok(vec![CursorUpdate { key: key.clone(), // the last page deletes the key in the same commit as its repos value: next2 unmodified lines }]) }) .await?; match listing { PdsListing::Complete => info!(%host, "listed repos of a newly followed PDS"), PdsListing::Complete => info!(%host, ?walk, "listed repos of a PDS"), PdsListing::Banned => { info!(%host, "PDS was banned, dropping its crawl"); self.forget(key).await?; }7 unmodified lines .await }}
fn load_queued(db: &crate::db::Db) -> Result<Vec<(String, Option<SmolStr>)>> { db.cursors .prefix(keys::NEW_PDS_CRAWL_PREFIX) .map(|entry| { let (key, value) = entry.into_inner().into_diagnostic()?; let host = std::str::from_utf8(&key[keys::NEW_PDS_CRAWL_PREFIX.len()..]) .into_diagnostic()? .to_owned(); Ok((host, new_pds_crawl::decode_cursor(&value)?))fn load_queued(db: &crate::db::Db) -> Result<Vec<(Walk, String, Option<SmolStr>)>> { Walk::ALL .into_iter() .flat_map(|walk| { db.cursors.prefix(walk.prefix()).map(move |entry| { let (key, value) = entry.into_inner().into_diagnostic()?; let host = std::str::from_utf8(&key[walk.prefix().len()..]) .into_diagnostic()? .to_owned(); Ok((walk, host, new_pds_crawl::decode_cursor(&value)?)) }) }) .collect()}
179 unmodified linesExpand all let state = test_state(&dir)?; let (addr, pds) = mock_pds().await?; let crawler = crawler(&state, addr)?; let url = Url::parse("http://pds.example/").into_diagnostic()?; let key = keys::new_pds_crawl_key("pds.example"); queue(&state, "pds.example", None)?;
pds.fail_second_page.store(true, Ordering::SeqCst); assert!( crawler .clone() .list_host("pds.example", key.clone(), &url, None) .list_host("pds.example", Walk::New, &url, None) .await .is_err() ); eventually(|| { Ok(tracked(&state, "did:web:first.example")? && load_queued(&state.db)? == vec![("pds.example".into(), Some("next".into()))]) && load_queued(&state.db)? == vec![(Walk::New, "pds.example".into(), Some("next".into()))]) }) .await?;
// the 500 put the PDS in backoff, which the producer waits out between rounds pds.fail_second_page.store(false, Ordering::SeqCst); state.throttler.get_handle(&url).await.record_success(); let [(host, cursor)] = <[_; 1]>::try_from(load_queued(&state.db)?).unwrap(); crawler.list_host(&host, key, &url, cursor).await?; let [(walk, host, cursor)] = <[_; 1]>::try_from(load_queued(&state.db)?).unwrap(); crawler.list_host(&host, walk, &url, cursor).await?; eventually(|| { Ok(tracked(&state, "did:web:second.example")? && load_queued(&state.db)?.is_empty()) }) .await?;29 unmodified lines ]; 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?; crawler .list_host("pds.example", Walk::New, &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(()) }
/// `did`'s metadata, tracked or not, with no backfill queued and `retry_count` failures so far fn metadata( state: &AppState, did: &'static str, tracked: bool, retry_count: u32, ) -> Result<()> { let mut metadata = crate::types::RepoMetadata::backfilling(rand::random()); metadata.tracked = tracked; metadata.retry_count = retry_count; state .db .repo_metadata .insert( keys::repo_metadata_key(&Did::new_static(did)?), crate::db::ser_repo_meta(&metadata)?, ) .into_diagnostic() }
fn set_status( state: &AppState, did: &'static str, status: crate::types::RepoStatus, ) -> Result<()> { let did = Did::new_static(did)?; let mut txn = crate::db::Txn::new(&state.db); let mut repo = txn.read_repo(&did)?.expect("the repo is stored"); repo.status = status; txn.put_repo(&did, &repo)?; txn.commit() }
#[tokio::test] async fn a_rewalk_resyncs_stuck_repos_and_leaves_synced_ones_to_the_firehose() -> Result<()> { use crate::types::{RepoStatus, ResyncErrorKind, ResyncState}; 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"); let here = "https://pds.example/"; for (did, pds) in [ ("did:web:moved.example", here), ("did:web:desynced.example", here), ("did:web:failed.example", here), ("did:web:untracked.example", here), ("did:web:elsewhere.example", "https://other.example/"), ] { synced_at(&state, did, old, pds)?; metadata(&state, did, did != "did:web:untracked.example", 4)?; } for did in [ "did:web:desynced.example", "did:web:untracked.example", "did:web:elsewhere.example", ] { set_status(&state, did, RepoStatus::Desynchronized)?; } let failed = keys::repo_key(&Did::new_static("did:web:failed.example")?); state .db .indexer .resync .insert( &failed, rmp_serde::to_vec(&ResyncState::Error { kind: ResyncErrorKind::Generic, retry_count: 4, next_retry: i64::MAX, }) .into_diagnostic()?, ) .into_diagnostic()?; *pds.known.lock().unwrap() = [ "did:web:moved.example", "did:web:desynced.example", "did:web:failed.example", "did:web:untracked.example", "did:web:elsewhere.example", ] .map(|did| json!({"did": did, "head": HEAD, "rev": new})) .to_vec(); state .db .cursors .insert( keys::rewalk_key("pds.example"), new_pds_crawl::encode_cursor(None)?, ) .into_diagnostic()?;
let url = Url::parse("http://pds.example/").into_diagnostic()?; crawler .list_host("pds.example", Walk::Rewalk, &url, None) .await?; eventually(|| { Ok(load_queued(&state.db)?.is_empty() && tracked(&state, "did:web:second.example")?) }) .await?; assert!(tracked(&state, "did:web:first.example")?); assert!(!backfill_queued(&state, "did:web:moved.example")?); assert!(backfill_queued(&state, "did:web:desynced.example")?); assert!(backfill_queued(&state, "did:web:failed.example")?); assert!(!backfill_queued(&state, "did:web:untracked.example")?); assert!(!backfill_queued(&state, "did:web:elsewhere.example")?); assert!( !state .db .indexer .resync .contains_key(&failed) .into_diagnostic()? ); let metadata = crate::db::deser_repo_meta( &state .db .repo_metadata .get(keys::repo_metadata_key(&Did::new_static( "did:web:failed.example", )?)) .into_diagnostic()? .expect("the failed repo keeps its metadata"), )?; assert_eq!(metadata.retry_count, 4, "a rewalk keeps the repo's backoff"); 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?;9 unmodified lines }); let url = Url::parse("http://pds.example/").into_diagnostic()?; crawler .clone() .list_host( "pds.example", keys::new_pds_crawl_key("pds.example"), &url, None, ) .list_host("pds.example", Walk::New, &url, None) .await?;
queue(&state, "gone.example", None)?; crawler.clone().crawl("gone.example", None).await?; crawler .clone() .crawl(Walk::New, "gone.example", None) .await?; queue(&state, "127.0.0.1", None)?; crawler.crawl("127.0.0.1", None).await?; crawler.crawl(Walk::New, "127.0.0.1", None).await?;
eventually(|| Ok(load_queued(&state.db)?.is_empty())).await?; assert_eq!(pds.requests.load(Ordering::SeqCst), 0); Ok(())2 unmodified lines11 unmodified lines121314151616171819202122232425266 unmodified lines3334353637383940414243444546474849506 unmodified lines575859604861626364656655676869707126 unmodified lines989910010190102103104105106107108109110103 unmodified linesExpand all21421521621720420520620721821922022122222322422522622722822923021423123223323423523623723823924024124224324424524624724824925025125225325425525625725825926026126226326426539 unmodified lines30530630730830931031131231331427427527631531627831731831932032132232332432532628632732832928929029133033133233333433533633733833934034134234334434529634634734834935035135235335435535635735830830931031131231331431531635936036136236336436536636736836937037137237337437537637737837938038138238338438538638738838939039139239339439532539639739839940049 unmodified lines4504514524533833843854544554564574581 unmodified line46046146246339546446546646746846961 unmodified lines11 unmodified lines
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::db::{Db, keys};use crate::ingest::indexer::message::RepoOp;use crate::ingest::indexer::repos::{on_owners, repo_ops};use crate::net::{PublicEndpoint, PublicHttpClient};use crate::state::AppState;use crate::types::{RepoState, RepoStatus, ResyncState};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);6 unmodified lines /// the host was banned locally partway through, so the rest was skipped. Banned,}
/// what a walk does with the known repos it lists whose doc points at the walked host.#[derive(Debug, Clone, Copy, PartialEq, Eq)]pub(crate) enum KnownRepos { Skip, /// resync the ones listed past the rev we hold ResyncNewer, /// resync the ones stuck desynchronized or failing their backfill ResyncStuck,}
/// 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,6 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, known: KnownRepos, progress: impl Fn(Option<&SmolStr>) -> Result<Vec<CursorUpdate>>, ) -> Result<PdsListing> { let hostname = pds .host_str()26 unmodified lines self.send_cursor_updates(progress(None)?).await?; return Ok(PdsListing::Complete); }; let page = self .parse_page(bytes, resync_newer.then(|| SmolStr::new(hostname))) .parse_page(bytes, known, SmolStr::new(hostname)) .await?; self.stats.record_crawled(page.count); // before the page's cursor moves. a walk cut short resyncs them next time self.resync(page.stuck).await?;
let state = &self.checker.state; let mut unknown = page.unknown_dids; let mut skipped = 0;103 unmodified linesExpand all throttle.record_success(); Ok(Some(bytes)) }
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(); async fn parse_page( &self, bytes: bytes::Bytes, known: KnownRepos, host: SmolStr, ) -> Result<Page> { let db = self.checker.state.db.clone(); tokio::time::timeout( BLOCKING_TASK_TIMEOUT, 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, resync_host.as_deref()) select_unknown_repos(&db, output, known, &host) }), ) .await .map_err(|_| miette::miette!("PDS listRepos parse timed out"))? .into_diagnostic()? }
/// the same resync an operator asks for, which keeps each repo's retry count async fn resync(&self, dids: Vec<Did>) -> Result<()> { if dids.is_empty() { return Ok(()); } let answers = on_owners( &self.checker.state, dids, |did| did, |state, shard, batch| repo_ops(state, shard.as_deref(), RepoOp::Resync, &batch), ) .await; let mut resynced = 0; for answer in answers { resynced += answer??.len(); } if resynced > 0 { debug!(resynced, "resyncing stuck repos a walk listed"); self.checker.state.notify_backfill(); } Ok(()) }
pub(super) async fn send_cursor_updates( &self, cursor_updates: Vec<CursorUpdate>, ) -> Result<()> {39 unmodified lines
struct Page { unknown_dids: Vec<Did>, newer: Vec<RepoListing>, stuck: Vec<Did>, cursor: Option<SmolStr>, count: usize,}
fn select_unknown_repos( repos: &fjall::Keyspace, filter: &fjall::Keyspace, crawler: &fjall::Keyspace, db: &Db, output: ListReposOutput, resync_host: Option<&str>, known: KnownRepos, host: &str,) -> Result<Page> { let count = output.repos.len(); let cursor = output.cursor; let mut unknown_dids = Vec::new(); let mut newer = Vec::new(); let mut stuck = Vec::new(); for repo in output.repos { let exclude_key = crate::db::filter::exclude_key(repo.did.as_str())?; if filter.contains_key(exclude_key).into_diagnostic()? { if db.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)?); if let Some(stored) = db.repos.get(keys::repo_key(&repo.did)).into_diagnostic()? { if known == KnownRepos::Skip { continue; } let stored = crate::db::deser_repo_state(&stored)?; if stored.pds.as_deref().and_then(pds_host).as_deref() != Some(host) { continue; } if known == KnownRepos::ResyncNewer { newer.extend(listing_if_newer(repo, &stored)); } else if is_stuck(db, &repo.did, &stored)? { stuck.push(repo.did.into_static()); } continue; } let retry_key = keys::crawler_retry_key(&repo.did); if !crawler.contains_key(retry_key).into_diagnostic()? { if !db.crawler.contains_key(retry_key).into_diagnostic()? { unknown_dids.push(repo.did); } } Ok(Page { unknown_dids, newer, stuck, 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);/// a tracked repo left desynchronized or with a failed backfill waiting on its retryfn is_stuck(db: &Db, did: &Did, stored: &RepoState<'_>) -> Result<bool> { let tracked = db .repo_metadata .get(keys::repo_metadata_key(did)) .into_diagnostic()? .map(|bytes| crate::db::deser_repo_meta(&bytes)) .transpose()? .is_some_and(|metadata| metadata.tracked); if !tracked { return Ok(false); } if matches!( stored.status, RepoStatus::Desynchronized | RepoStatus::Error(_) ) { return Ok(true); } let failed = db .indexer .resync .get(keys::repo_key(did)) .into_diagnostic()? .map(|bytes| rmp_serde::from_slice::<ResyncState>(&bytes).into_diagnostic()) .transpose()?; Ok(matches!(failed, Some(ResyncState::Error { .. })))}
/// `repo` as a listing if it's listed past the rev we holdfn listing_if_newer(repo: ListedRepo<'_>, stored: &RepoState<'_>) -> Option<RepoListing> { 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)) listing.claims_newer(stored).then_some(listing)}
#[cfg(test)]mod tests {49 unmodified lines .insert(keys::crawler_retry_key(&retry), b"") .into_diagnostic()?;
let page = select_unknown_repos( &state.db.repos, &state.db.filter, &state.db.crawler, &state.db, list_repos_output(json!({ "cursor": "next", "repos": [ {"did": known, "head": HEAD, "rev": "3j6s746574657", "active": false, "status": "takendown"},1 unmodified line {"did": retry, "head": HEAD, "rev": "3j6s746574657"}, {"did": "did:web:inactive-unknown.example", "head": HEAD, "rev": "3j6s746574657", "active": false, "status": "deactivated"} ] }))?, None, KnownRepos::Skip, "pds.example", )?;
assert_eq!(page.count, 4); assert_eq!(page.cursor.as_deref(), Some("next"));61 unmodified lines10 unmodified lines11121314151516171819279 unmodified linesExpand all299300301302303303304305306307339 unmodified linesExpand all10 unmodified linesuse tracing::{debug, error, info, trace, warn};use url::Url;
use super::list_repos::ListReposProducer;use super::pds_repos::PdsRepoLister;use super::pds_repos::{KnownRepos, PdsRepoLister};use super::worker::{CrawlerBatch, CursorUpdate};use super::{CrawlerStats, InFlight, SignalChecker, base_url};use crate::db::keys;use crate::net::{PublicHost, PublicHttpClient, public_http, trusted_http};279 unmodified linesExpand all return Ok(()); } let start = saved.and_then(|state| state.cursor); self.lister .list(pds, account_count, http, start, false, |next| { .list(pds, account_count, http, start, KnownRepos::Skip, |next| { let state = next.map_or_else( || PassCursor::complete(pass), |next| PassCursor::active(pass, Some(next.clone())), );339 unmodified linesExpand all330 unmodified linesExpand all331332333334335336337338339340341342343344345346347348349350351352353354355356357358283 unmodified linesExpand all330 unmodified linesExpand all key.extend_from_slice(host.as_bytes()); key}
/// key format: `rewalk|<host>`, a `listRepos` walk a followed PDS asked for/// through `requestCrawl`. the value is its page cursor.pub const REWALK_PREFIX: &[u8] = b"rewalk|";
pub fn rewalk_key(host: &str) -> Vec<u8> { let mut key = Vec::with_capacity(REWALK_PREFIX.len() + host.len()); key.extend_from_slice(REWALK_PREFIX); key.extend_from_slice(host.as_bytes()); key}
/// key format: `rewalk_allowance|<host>`, the rewalks a PDS used in its current window.pub fn rewalk_allowance_key(host: &str) -> Vec<u8> { let prefix = b"rewalk_allowance|"; let mut key = Vec::with_capacity(prefix.len() + host.len()); key.extend_from_slice(prefix); key.extend_from_slice(host.as_bytes()); key}
/// key format: `seed_listed|<seed url>`, set once a seed relay's `listHosts`/// has been read to the end with every host on it admitted.pub const SEED_LISTED_PREFIX: &[u8] = b"seed_listed|";
283 unmodified linesExpand all