use std::time::Duration;
use miette::{IntoDiagnostic, Result};
use smol_str::SmolStr;
use tokio::task::JoinSet;
use tracing::{debug, error, info, warn};
use url::Url;
use super::pds_repos::{KnownRepos, PdsListing, PdsRepoLister};
use super::worker::CursorUpdate;
use crate::control::new_pds_crawl::{self, Budget};
use crate::db::keys;
use crate::net::{PublicHost, PublicHttpClient};
use crate::util::WatchEnabledExt;
const MAX_CONCURRENT_CRAWLS: usize = 4;
const RETRY_DELAY: Duration = Duration::from_secs(5 * 60);
/// 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::ResyncNewerOrStuck,
}
}
}
/// how one host's walk went this round.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum Walked {
Done,
/// a rewalk that spent its bytes, until the unix time its window ends
WaitsUntil(i64),
}
/// what a round of [`NewPdsCrawler::crawl_queued`] leaves for the next one.
#[derive(Debug, Default)]
struct Round {
failed: bool,
resume_at: Option<i64>,
}
impl Round {
/// how long to sleep before the next round, `None` to wait for a wake
fn delay(&self, now: i64) -> Option<Duration> {
let waits = self
.resume_at
.map(|at| Duration::from_secs((at - now).max(1) as u64));
self.failed
.then_some(RETRY_DELAY)
.into_iter()
.chain(waits)
.min()
}
fn settle(
&mut self,
result: std::result::Result<(String, Result<Walked>), tokio::task::JoinError>,
) {
match result {
Ok((_, Ok(Walked::Done))) => {}
Ok((_, Ok(Walked::WaitsUntil(at)))) => {
self.resume_at = Some(self.resume_at.map_or(at, |soonest| soonest.min(at)));
}
Ok((host, Err(err))) => {
warn!(%host, err = ?err, "new PDS crawl failed, retrying later");
self.failed = true;
}
Err(err) => {
error!(err = ?err, "new PDS crawl task failed");
self.failed = true;
}
}
}
}
/// 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,
}
impl NewPdsCrawler {
pub(crate) async fn run(mut self) {
loop {
self.lister
.enabled
.wait_enabled(self.lister.component)
.await;
let round = self.crawl_queued().await.unwrap_or_else(|err| {
error!(err = ?err, "failed to read queued new PDS crawls");
Round {
failed: true,
resume_at: None,
}
});
// a wake that lands mid round is kept by the notify, so the next round sees its host
let state = self.lister.checker.state.clone();
let woken = state.new_pds_crawl_notify.notified();
match round.delay(chrono::Utc::now().timestamp()) {
None => woken.await,
Some(delay) => tokio::select! {
_ = woken => {}
_ = tokio::time::sleep(delay) => {}
},
}
}
}
async fn crawl_queued(&self) -> Result<Round> {
let queued = self.lister.checker.state.db.run(load_queued).await?;
let mut round = Round::default();
let mut crawls = JoinSet::new();
for (walk, host, cursor) in queued {
if crawls.len() >= MAX_CONCURRENT_CRAWLS
&& let Some(result) = crawls.join_next().await
{
round.settle(result);
}
let crawler = self.clone();
crawls.spawn(async move {
let result = crawler.crawl(walk, &host, cursor).await;
(host, result)
});
}
while let Some(result) = crawls.join_next().await {
round.settle(result);
}
Ok(round)
}
async fn crawl(self, walk: Walk, host: &str, cursor: Option<SmolStr>) -> Result<Walked> {
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");
return self.forget(key).await.map(|()| Walked::Done);
}
};
// an operator removing the source means they don't want this host either
let source_key = keys::firehose_source_key(public.firehose_url().as_str());
let followed = self
.lister
.checker
.state
.db
.run(move |db| db.crawler.contains_key(source_key).into_diagnostic())
.await?;
if !followed {
info!(%host, "PDS is no longer followed, dropping its crawl");
return self.forget(key).await.map(|()| Walked::Done);
}
let budget = self.rewalk_budget(walk, host).await?;
if let Some(until) = budget
.as_ref()
.and_then(|budget| budget.spent_until(chrono::Utc::now().timestamp()))
{
debug!(%host, until, "rewalk waits for its window to end");
return Ok(Walked::WaitsUntil(until));
}
info!(%host, ?walk, resuming = cursor.is_some(), "listing repos of a PDS");
let pds = public.https_endpoint().url().clone();
self.list_host(host, walk, &pds, cursor, budget).await
}
/// a rewalk is held to the host's allowance, a new host's first walk isn't
async fn rewalk_budget(&self, walk: Walk, host: &str) -> Result<Option<Budget>> {
if walk != Walk::Rewalk {
return Ok(None);
}
let state = &self.lister.checker.state;
let (limit, window) = (state.rewalk_bytes, state.rewalk_window.as_secs() as i64);
let host = host.to_owned();
state
.db
.run(move |db| Budget::load(db, &host, limit, window).map(Some))
.await
}
async fn list_host(
mut self,
host: &str,
walk: Walk,
pds: &Url,
cursor: Option<SmolStr>,
mut budget: Option<Budget>,
) -> Result<Walked> {
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,
walk.known_repos(),
budget.as_mut(),
|next| {
Ok(vec![CursorUpdate {
key: key.clone(),
// the last page deletes the key in the same commit as its repos
value: next
.map(|next| new_pds_crawl::encode_cursor(Some(next)))
.transpose()?,
}])
},
)
.await?;
match listing {
PdsListing::Complete => info!(%host, ?walk, "listed repos of a PDS"),
PdsListing::Banned => {
info!(%host, "PDS was banned, dropping its crawl");
self.forget(key).await?;
}
PdsListing::OutOfBudget { until } => {
info!(%host, until, "rewalk spent its bytes, resuming when its window ends");
return Ok(Walked::WaitsUntil(until));
}
}
Ok(Walked::Done)
}
async fn forget(&self, key: Vec<u8>) -> Result<()> {
self.lister
.send_cursor_updates(vec![CursorUpdate { key, value: None }])
.await
}
}
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()
}
#[cfg(test)]
mod tests {
use super::*;
use crate::config::Config;
use crate::crawler::{CrawlerStats, CrawlerWorker, InFlight, SignalChecker};
use crate::net::PublicHost;
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, Tid};
use serde_json::json;
use std::collections::HashMap;
use std::sync::Arc;
use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering};
use tempfile::tempdir;
const HEAD: &str = "bafyreidf747c4x3lps3k4n357l3a3r57k3k465743k573k465743k5";
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, 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") {
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();
}
Json(json!({
"repos": [{"did": "did:web:second.example", "head": HEAD, "rev": "3j6s746574657"}]
}))
.into_response()
}
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());
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 });
Ok((addr, pds))
}
/// a crawler whose `pds.example` points at `addr`, with a running worker.
fn crawler(state: &Arc<AppState>, addr: std::net::SocketAddr) -> Result<NewPdsCrawler> {
let stats = CrawlerStats::new(state.clone(), Vec::new(), state.throttler.clone());
let (worker, tx) = CrawlerWorker::new(state.clone(), 100, 50, stats.clone());
tokio::spawn(worker.run());
Ok(NewPdsCrawler {
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: "new PDS crawler",
},
http: PublicHttpClient::test(
reqwest::Client::builder()
.resolve("pds.example", addr)
.build()
.into_diagnostic()?,
),
})
}
fn test_state(dir: &tempfile::TempDir) -> Result<Arc<AppState>> {
Ok(Arc::new(AppState::new(&Config {
database_path: dir.path().to_path_buf(),
full_network: true,
enable_crawler: Some(true),
..Default::default()
})?))
}
fn queue(state: &AppState, host: &str, cursor: Option<&str>) -> Result<()> {
let cursor = cursor.map(SmolStr::new);
state
.db
.cursors
.insert(
keys::new_pds_crawl_key(host),
new_pds_crawl::encode_cursor(cursor.as_ref())?,
)
.into_diagnostic()
}
fn tracked(state: &AppState, did: &'static str) -> Result<bool> {
state
.db
.repos
.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()? {
tokio::time::sleep(Duration::from_millis(10)).await;
}
Ok(())
})
.await
.into_diagnostic()?
}
#[tokio::test]
async fn a_broken_walk_keeps_its_cursor_and_the_last_page_clears_it() -> Result<()> {
let dir = tempdir().into_diagnostic()?;
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()?;
queue(&state, "pds.example", None)?;
pds.fail_second_page.store(true, Ordering::SeqCst);
assert!(
crawler
.clone()
.list_host("pds.example", Walk::New, &url, None, None)
.await
.is_err()
);
eventually(|| {
Ok(tracked(&state, "did:web:first.example")?
&& 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 [(walk, host, cursor)] = <[_; 1]>::try_from(load_queued(&state.db)?).unwrap();
crawler.list_host(&host, walk, &url, cursor, None).await?;
eventually(|| {
Ok(tracked(&state, "did:web:second.example")? && load_queued(&state.db)?.is_empty())
})
.await?;
// first page, the failed second page, then the second page again
assert_eq!(pds.requests.load(Ordering::SeqCst), 3);
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, None)
.await;
assert!(format!("{:?}", result.unwrap_err()).contains("too large"));
assert!(!tracked(&state, "did:web:first.example")?);
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, None)
.await;
assert!(format!("{:?}", result.unwrap_err()).contains("already returned"));
assert_eq!(requests.load(Ordering::SeqCst), 3);
Ok(())
}
fn spent_allowance(state: &AppState, since: i64) -> Result<()> {
state
.db
.cursors
.insert(
keys::rewalk_allowance_key("pds.example"),
rmp_serde::to_vec(&Some(crate::types::Allowance {
since,
used: u32::MAX,
}))
.into_diagnostic()?,
)
.into_diagnostic()
}
#[tokio::test]
async fn a_rewalk_that_spends_its_bytes_keeps_its_cursor_until_the_window_ends() -> Result<()> {
let dir = tempdir().into_diagnostic()?;
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 window = state.rewalk_window.as_secs() as i64;
let budget = |bytes| Budget::load(&state.db, "pds.example", bytes, window);
state
.db
.cursors
.insert(
keys::rewalk_key("pds.example"),
new_pds_crawl::encode_cursor(None)?,
)
.into_diagnostic()?;
// the first page overshoots one byte, so the walk stops before the second
let walked = crawler
.clone()
.list_host("pds.example", Walk::Rewalk, &url, None, Some(budget(1)?))
.await?;
let Walked::WaitsUntil(until) = walked else {
panic!("the walk should wait for its window, not {walked:?}");
};
eventually(|| {
Ok(tracked(&state, "did:web:first.example")?
&& load_queued(&state.db)?
== vec![(Walk::Rewalk, "pds.example".into(), Some("next".into()))])
})
.await?;
let spent = budget(1)?;
assert_eq!(spent.spent_until(until - 1), Some(until));
assert_eq!(pds.requests.load(Ordering::SeqCst), 1);
assert!(!tracked(&state, "did:web:second.example")?);
// it restarted with its window still running: nothing is fetched, the cursor stays
spent_allowance(&state, until - window)?;
let followed = PublicHost::parse("pds.example").into_diagnostic()?;
state
.db
.crawler
.insert(
keys::firehose_source_key(followed.firehose_url().as_str()),
b"",
)
.into_diagnostic()?;
let walked = crawler
.clone()
.crawl(Walk::Rewalk, "pds.example", Some("next".into()))
.await?;
assert!(matches!(walked, Walked::WaitsUntil(_)));
assert_eq!(pds.requests.load(Ordering::SeqCst), 1);
// the window ended: the walk goes on from its cursor and clears it
spent_allowance(&state, 0)?;
state.throttler.get_handle(&url).await.record_success();
let walked = crawler
.list_host(
"pds.example",
Walk::Rewalk,
&url,
Some("next".into()),
Some(budget(1 << 20)?),
)
.await?;
assert_eq!(walked, Walked::Done);
eventually(|| {
Ok(tracked(&state, "did:web:second.example")? && load_queued(&state.db)?.is_empty())
})
.await?;
assert_eq!(pds.requests.load(Ordering::SeqCst), 2);
Ok(())
}
#[test]
fn a_round_sleeps_until_the_soonest_retry_or_window_end() {
let round = |failed, resume_at| Round { failed, resume_at };
assert_eq!(round(false, None).delay(100), None);
assert_eq!(round(true, None).delay(100), Some(RETRY_DELAY));
assert_eq!(
round(false, Some(160)).delay(100),
Some(Duration::from_secs(60))
);
assert_eq!(
round(true, Some(160)).delay(100),
Some(Duration::from_secs(60))
);
assert_eq!(round(true, Some(100 + 3600)).delay(100), Some(RETRY_DELAY));
assert_eq!(
round(false, Some(90)).delay(100),
Some(Duration::from_secs(1)),
"a window that ended since the round started can't spin"
);
}
#[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()?;
crawler
.list_host("pds.example", Walk::New, &url, None, 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_ones_listed_at_a_newer_rev() -> 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:stuckathead.example", here),
("did:web:stuckfuture.example", here),
("did:web:failed.example", here),
("did:web:untracked.example", here),
("did:web:elsewhere.example", "https://other.example/"),
] {
let rev = if did == "did:web:stuckathead.example" {
new
} else {
old
};
synced_at(&state, did, rev, pds)?;
metadata(&state, did, did != "did:web:untracked.example", 4)?;
}
for did in [
"did:web:desynced.example",
"did:web:stuckathead.example",
"did:web:stuckfuture.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:stuckathead.example",
"did:web:stuckfuture.example",
"did:web:failed.example",
"did:web:untracked.example",
"did:web:elsewhere.example",
]
.map(|did| {
let rev = if did == "did:web:stuckfuture.example" {
"jzzzzzzzzzzzz"
} else {
new
};
json!({"did": did, "head": HEAD, "rev": rev})
})
.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, 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:stuckathead.example")?,
"a stuck repo listed at the rev we hold is no newer, but is still resynced"
);
assert!(
backfill_queued(&state, "did:web:stuckfuture.example")?,
"a stuck repo listed at a rev from the future is still resynced, not dropped as news"
);
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?;
let crawler = crawler(&state, addr)?;
queue(&state, "pds.example", None)?;
state.pds_meta.rcu(|meta| {
let mut next = (**meta).clone();
next.update_host_entry("pds.example", |entry| {
entry.status = crate::pds_meta::HostStatus::Banned;
});
next
});
let url = Url::parse("http://pds.example/").into_diagnostic()?;
crawler
.clone()
.list_host("pds.example", Walk::New, &url, None, None)
.await?;
queue(&state, "gone.example", None)?;
crawler
.clone()
.crawl(Walk::New, "gone.example", None)
.await?;
queue(&state, "127.0.0.1", None)?;
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(())
}
}