[crawler] don't trust listed revs from the future
196 unmodified linesExpand all197198199200201201202203204205206698 unmodified linesExpand all196 unmodified linesExpand all /// run MST inversion on live commits. `SignatureVerification::Full` already does that for every /// commit, so this only matters with `backfill-only` or `none`. set via `HYDRANT_VERIFY_MST`. pub verify_mst: bool, /// clock drift window for future-rev rejection, in seconds. /// commits with a rev timestamp more than this many seconds in the future are rejected. /// a rev further ahead of our clock than this gets rejected in a commit and doesn't count as /// newer in a listing. /// set via `HYDRANT_REV_CLOCK_SKEW`. default: 300 (5 minutes). pub rev_clock_skew_secs: i64,
/// NSID patterns that trigger auto-discovery in filter mode (e.g. `app.bsky.feed.post`).698 unmodified linesExpand all82 unmodified lines83848586878889909192139 unmodified linesExpand all232233234235236237238239240241410 unmodified linesExpand all82 unmodified lines /// 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(crate) rev_clock_skew_secs: i64, #[cfg(feature = "indexer")] pub crawler_enabled: watch::Sender<bool>, #[cfg(feature = "indexer")] pub backfill_enabled: watch::Sender<bool>, #[cfg(feature = "indexer")]139 unmodified linesExpand all rewalk_window: config .offline_host_retry_interval .unwrap_or(crate::control::new_pds_crawl::DEFAULT_REWALK_WINDOW), #[cfg(feature = "indexer")] rev_clock_skew_secs: config.rev_clock_skew_secs, #[cfg(feature = "indexer")] crawler_enabled, firehose_enabled, #[cfg(feature = "indexer")] backfill_enabled,410 unmodified linesExpand all220 unmodified linesExpand all22122222322422522622722822923023123123223323423523679 unmodified lines31631731831932032132232332412 unmodified lines33733833934033934134234334434540 unmodified lines38638738838938839039139239339439539639739839940040139640240340440540622 unmodified lines42943043143243343443543643743843944044144244344444544644744844945045145245345445545645745845946046146246346446531 unmodified lines49749849950050150250350450561 unmodified lines220 unmodified linesExpand all known: KnownRepos, host: SmolStr, ) -> Result<Page> { let db = self.checker.state.db.clone(); let skew_secs = self.checker.state.rev_clock_skew_secs; 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(&db, output, known, &host) select_unknown_repos(&db, output, known, &host, skew_secs) }), ) .await .map_err(|_| miette::miette!("PDS listRepos parse timed out"))?79 unmodified lines db: &Db, output: ListReposOutput, known: KnownRepos, host: &str, skew_secs: i64,) -> Result<Page> { let count = output.repos.len(); let cursor = output.cursor; let mut unknown_dids = Vec::new();12 unmodified lines 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)); newer.extend(listing_if_newer(repo, &stored, skew_secs)); } else if is_stuck(db, &repo.did, &stored)? { stuck.push(repo.did.into_static()); } continue;40 unmodified lines 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> {fn listing_if_newer( repo: ListedRepo<'_>, stored: &RepoState<'_>, skew_secs: i64,) -> 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, }; listing.claims_newer(stored).then_some(listing) listing.claims_newer(stored, skew_secs).then_some(listing)}
#[cfg(test)]mod tests {22 unmodified lines }) }
#[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(), ..Default::default() })?; let did = Did::new_static("did:web:alice.example").into_diagnostic()?; let mut row = RepoState::synced(); row.pds = Some(jacquard_common::CowStr::Borrowed("https://pds.example/")); let mut txn = crate::db::Txn::new(&state.db); txn.put_repo(&did, &row)?; txn.commit()?;
for (rev, listed) in [("3mz2j7sg2kh22", 1), ("jzzzzzzzzzzzz", 0)] { let page = select_unknown_repos( &state.db, list_repos_output(json!({ "repos": [{"did": did.as_str(), "head": HEAD, "rev": rev}] }))?, KnownRepos::ResyncNewer, "pds.example", state.rev_clock_skew_secs, )?; assert_eq!(page.newer.len(), listed, "{rev}"); } Ok(()) }
#[test] fn pages_select_only_unknown_repos_without_trusting_status() -> Result<()> { let tmp = tempdir().into_diagnostic()?; let config = Config { database_path: tmp.path().to_path_buf(),31 unmodified lines ] }))?, KnownRepos::Skip, "pds.example", state.rev_clock_skew_secs, )?;
assert_eq!(page.count, 4); assert_eq!(page.cursor.as_deref(), Some("next"));61 unmodified lines2 unmodified lines3456789101161 unmodified lines737475767677787778798081828384504 unmodified linesExpand all58959059159259259359459559659755 unmodified lines6536546556566576586596606616626636646656646666676686696706696706716716726736746756766773 unmodified lines6816826836846836856866876886896906896906916926936946956963 unmodified lines7007017027037027047057067077067077087097087097107117127137147153 unmodified lines719720721722721723724725726727728729728730731732733734735734735736737738739736737738739740741742743744745112 unmodified linesExpand all85885986086186286386486586656 unmodified lines92392492592692792892993093193293393493593692493793893994094194293093194394494594694720 unmodified lines968969970971960972973962974975976977978561 unmodified linesExpand all15401541154215431544154515461547154815491550155115521553155415551556155715581559156015611562156315641565156615671568310 unmodified linesExpand all2 unmodified linesuse crate::ingest::indexer::IndexerMessage;#[cfg(test)]use crate::ingest::indexer::message::shard_of;use crate::ingest::indexer::repos::on_owners;use crate::ingest::validation::is_future_rev;use crate::state::AppState;use crate::types::{BulkJobV1, GaugeState, RepoMetadata, RepoState, RepoStatus, ResyncState};pub(crate) type ListedRepoStatus = crate::ingest::stream::AccountStatus<'static>;use jacquard_common::IntoStatic;61 unmodified lines fn starts_a_wipe(&self) -> bool { !self.active && listed_repo_status(self.status.as_ref()) == RepoStatus::Deleted }
/// whether the listing claims a newer rev than `stored` has, the one thing a listing says /// about events hydrant may not have seen yet pub(crate) fn claims_newer(&self, stored: &RepoState<'_>) -> bool { /// a rev more than `skew_secs` ahead of our clock never counts as newer pub(crate) fn claims_newer(&self, stored: &RepoState<'_>, skew_secs: i64) -> bool { self.active && !is_future_rev(&self.rev.to_tid(), skew_secs) && stored .root .as_ref() .is_none_or(|root| self.rev.as_bytes() > root.rev.as_bytes())504 unmodified linesExpand all let db = &state.db; let mut applied = Applied::default(); for listing in listings { let mut txn = crate::db::Txn::new(db); let outcome = reconcile_listing(&mut txn, listing)?; let outcome = reconcile_listing(state, &mut txn, listing)?; txn.commit()?; if outcome != ReconcileOutcome::Unchanged { applied.reconciled += 1; if listing.starts_a_wipe() {55 unmodified lines }}
pub(crate) fn reconcile_listing( state: &AppState, txn: &mut crate::db::Txn<'_>, listing: &RepoListing,) -> Result<ReconcileOutcome> { let db = txn.db; if txn.is_excluded(&listing.did)? { return Ok(ReconcileOutcome::Unchanged); } let did_key = keys::repo_key(&listing.did); let Some(mut state) = txn.read_repo(&listing.did)? else { let Some(mut repo) = txn.read_repo(&listing.did)? else { if listing.active || !listing.create_if_missing { return Ok(ReconcileOutcome::Unchanged); } let status = listed_repo_status(listing.status.as_ref()); let mut state = RepoState::backfilling(); state.active = false; state.status = status.clone(); let mut repo = RepoState::backfilling(); repo.active = false; repo.status = status.clone(); if status != RepoStatus::Deleted { txn.batch.insert( &db.indexer.resync, &did_key,3 unmodified lines .into_diagnostic()?, ); txn.transition_lifecycle(&listing.did, GaugeState::Resync(None))?; } txn.put_repo(&listing.did, &state)?; txn.put_repo(&listing.did, &repo)?; txn.counts.add_repos(1); return Ok(ReconcileOutcome::StateOnly); };
if listing.active { let listing_is_newer = listing.claims_newer(&state); if state.active && !listing_is_newer { let listing_is_newer = listing.claims_newer(&repo, state.rev_clock_skew_secs); if repo.active && !listing_is_newer { return Ok(ReconcileOutcome::Unchanged); } if !listing_is_newer { // the inactive status can be the pds's own answer, so a relay saying active3 unmodified lines if listing_is_pending(db, &listing.did)? || !listing_may_queue(db, &listing.did)? { return Ok(ReconcileOutcome::Unchanged); }
let was_active = state.active; let was_active = repo.active; txn.batch.remove(&db.indexer.resync, &did_key); enqueue_listing_backfill(txn, &listing.did)?; txn.transition_lifecycle(&listing.did, GaugeState::Pending)?; state.status = RepoStatus::Desynchronized; state.active = true; state.touch(); txn.put_repo(&listing.did, &state)?; repo.status = RepoStatus::Desynchronized; repo.active = true; repo.touch(); txn.put_repo(&listing.did, &repo)?; return Ok(if was_active { ReconcileOutcome::StateOnly } else { ReconcileOutcome::AccountChanged3 unmodified lines let status = listed_repo_status(listing.status.as_ref()); if status != RepoStatus::Deleted { // a relay can take a repo down while its pds still serves it, and several relays can // disagree, so only the pds's own answer makes the account inactive return if state.active { return if repo.active { ask_pds(txn, &listing.did, status) } else { Ok(ReconcileOutcome::Unchanged) }; }
let account_changed = state.active || state.status != status; let account_changed = repo.active || repo.status != status; if !account_changed && !listing_is_pending(db, &listing.did)? { return Ok(ReconcileOutcome::Unchanged); }
txn.transition_lifecycle(&listing.did, GaugeState::Synced)?; crate::ops::start_wipe(&mut *txn, db, &listing.did, &state)?; state.root = None; state.active = false; state.status = status; state.touch(); txn.put_repo(&listing.did, &state)?; crate::ops::start_wipe(&mut *txn, db, &listing.did, &repo)?; repo.root = None; repo.active = false; repo.status = status; repo.touch(); txn.put_repo(&listing.did, &repo)?; Ok(if account_changed { ReconcileOutcome::AccountChanged } else { ReconcileOutcome::StateOnly112 unmodified linesExpand all#[cfg(test)]mod tests { use super::*; use bytes::Bytes; use jacquard_common::types::string::Tid; use miette::IntoDiagnostic;
#[test] fn pending_between_resume_and_max_does_not_start_throttling() {56 unmodified lines }); state }
struct TestDb(AppState);
impl std::ops::Deref for TestDb { type Target = crate::db::Db;
fn deref(&self) -> &Self::Target { &self.0.db } }
fn open_db() -> Result<(tempfile::TempDir, crate::db::Db)> { fn open_db() -> Result<(tempfile::TempDir, TestDb)> { let tmp = tempfile::tempdir().into_diagnostic()?; let config = crate::config::Config { database_path: tmp.path().to_path_buf(), ..Default::default() }; let db = crate::db::Db::open(&config)?; Ok((tmp, db)) Ok((tmp, TestDb(AppState::new(&config)?))) }
fn insert_repo( db: &crate::db::Db,20 unmodified lines } Ok(()) }
fn apply(db: &crate::db::Db, listing: &RepoListing) -> Result<ReconcileOutcome> { fn apply(db: &TestDb, listing: &RepoListing) -> Result<ReconcileOutcome> { let mut txn = crate::db::Txn::new(db); let changed = reconcile_listing(&mut txn, listing)?; let changed = reconcile_listing(&db.0, &mut txn, listing)?; txn.commit()?; Ok(changed) }
561 unmodified linesExpand all Ok(()) }
#[test] fn a_listed_rev_from_the_future_is_not_news() -> Result<()> { let (_tmp, db) = open_db()?; insert_repo(&db, &state_with_root(rev(1)), 7, false)?; let listed_at = |rev: &str| RepoListing::active(did(), DbTid::from(&Tid::new(rev).unwrap()));
assert_eq!( apply(&db, &listed_at("jzzzzzzzzzzzz"))?, ReconcileOutcome::Unchanged ); assert_eq!(db.indexer.pending_iter().count(), 0); let now = Tid::now_0(); assert_eq!( apply(&db, &listed_at(now.as_str()))?, ReconcileOutcome::StateOnly ); assert_eq!(db.indexer.pending_iter().count(), 1); Ok(()) }
#[test] fn terminal_cursor_update_clears_resume_point() -> Result<()> { let (_tmp, db) = open_db()?; let key = b"crawler_cursor|relay".to_vec(); db.cursors.insert(&key, b"old").into_diagnostic()?;310 unmodified linesExpand all123456181 unmodified linesExpand all188189190191191192193194195192193196194195196197112 unmodified linesExpand all310311312313314315316317318319320321322187 unmodified linesExpand alluse jacquard_common::types::crypto::PublicKey;use jacquard_common::types::tid::Tid;use miette::IntoDiagnostic;use thiserror::Error;
use crate::ingest::stream::{Commit, RepoOpAction, Sync};181 unmodified linesExpand all return Err(CommitValidationError::StaleRev); }
// 3. future rev, reject if rev timestamp is more than clock_skew_secs ahead of now { let rev_us = msg.rev.timestamp() as i64; let now_us = chrono::Utc::now().timestamp_micros(); if rev_us > now_us + opts.rev_clock_skew_secs * 1_000_000 { return Err(CommitValidationError::FutureRev); if is_future_rev(&msg.rev, opts.rev_clock_skew_secs) { return Err(CommitValidationError::FutureRev); } }
// 4. CAR parse, checking each block's cid as it is read let parsed = crate::car::parse_commit_car(msg.blocks.clone(), opts.verify_cids)112 unmodified linesExpand all
Ok(ValidatedSync { commit_obj })}
pub(crate) fn is_future_rev(rev: &Tid, skew_secs: i64) -> bool { let rev_us = rev.timestamp() as i64; rev_us > chrono::Utc::now().timestamp_micros() + skew_secs * 1_000_000}
/// whether `msg` picks up where `root` left off, as far as it says. a missing prevData isn't held/// against it here, unlike in [`breaks_chain`], since this asks again about a commit that went/// through that check when it came in#[cfg(feature = "indexer")]187 unmodified linesExpand all