[crawler] resync repos a rewalk finds at a newer rev
208 unmodified linesExpand all209210211212213213214215216217807 unmodified linesExpand all208 unmodified linesExpand allcodec-call:src/crawler/list_repos/producer.rs::pass_cursor_updates::.to_be_bytes rust:src/crawler/list_repos/producer.rs::inactive_unknown_repo_is_not_backfilledcodec-call:src/crawler/list_repos/producer.rs::pass_cursor_updates::rmp_serde::to_vec rust:src/crawler/list_repos/producer.rs::inactive_unknown_repo_is_not_backfilledcodec-call:src/crawler/list_repos/retry.rs::RetryProducer::process_queue::rmp_serde::from_slice nu:tests/crawler_full_network.nucodec-call:src/crawler/pds_repos.rs::PdsRepoLister::parse_page::serde_json::from_slice rust:src/crawler/new_pds.rs::a_broken_walk_keeps_its_cursor_and_the_last_page_clears_itcodec-call:src/crawler/pds_repos.rs::is_stuck::rmp_serde::from_slice rust:src/crawler/new_pds.rs::a_rewalk_resyncs_stuck_repos_and_leaves_synced_ones_to_the_firehosecodec-call:src/crawler/pds_repos.rs::is_stuck::rmp_serde::from_slice rust:src/crawler/new_pds.rs::a_rewalk_resyncs_stuck_repos_and_ones_listed_at_a_newer_revcodec-call:src/crawler/relay_first.rs::DeepDiscoveryProducer::crawl_pass::serde_json::from_slice rust:src/crawler/relay_first.rs::pass_cursor_round_tripscodec-call:src/crawler/relay_first.rs::DeepDiscoveryProducer::load_cursor::rmp_serde::from_slice rust:src/crawler/relay_first.rs::pass_cursor_round_tripscodec-call:src/crawler/relay_first.rs::read_u64::u64::from_be_bytes rust:src/crawler/relay_first.rs::pass_markers_are_read_big_endiancodec-call:src/crawler/relay_first.rs::serialize_cursor::rmp_serde::to_vec rust:src/crawler/relay_first.rs::pass_cursor_round_trips807 unmodified linesExpand all6 unmodified lines7891011121311121314151617207 unmodified linesExpand all6 unmodified lines//! `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.//! its repos listed at a newer rev than ours, or stuck desynchronized or failing.//! a host gets [`REWALKS_PER_WINDOW`] per rewalk window, and asking while one is//! queued or running does nothing.
#[cfg(feature = "indexer")]mod enabled { use miette::{IntoDiagnostic, Result};207 unmodified linesExpand all41 unmodified lines42434445464647484950421 unmodified linesExpand all4724734744754764764774784794802 unmodified lines48348448548648748848949049149249349449549649749149849950050150250350450550650750815 unmodified lines52452552652752852953053153253352353453553653753853954054154254354454511 unmodified lines55755855956054356156256356456556656756856957057157257357457554 unmodified lines41 unmodified lines
fn known_repos(self) -> KnownRepos { match self { Self::New => KnownRepos::ResyncNewer, Self::Rewalk => KnownRepos::ResyncStuck, Self::Rewalk => KnownRepos::ResyncNewerOrStuck, } }}
421 unmodified linesExpand all txn.commit() }
#[tokio::test] async fn a_rewalk_resyncs_stuck_repos_and_leaves_synced_ones_to_the_firehose() -> Result<()> { 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?;2 unmodified lines 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, old, pds)?; 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)?;15 unmodified lines .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| json!({"did": did, "head": HEAD, "rev": new})) .map(|did| { let rev = if did == "did:web:stuckfuture.example" { "jzzzzzzzzzzzz" } else { new }; json!({"did": did, "head": HEAD, "rev": rev}) }) .to_vec(); state .db .cursors11 unmodified lines 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: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!( !state54 unmodified lines38 unmodified lines394041424344434445464748287 unmodified linesExpand all33633733833934034134234034134234334434534640 unmodified lines387388389390391391392393394395396396397398399399400401402403183 unmodified linesExpand all38 unmodified linespub(crate) enum KnownRepos { Skip, /// resync the ones listed past the rev we hold ResyncNewer, /// resync the ones stuck desynchronized or failing their backfill ResyncStuck, /// stuck repos are resynced even when listed at the rev we hold ResyncNewerOrStuck,}
/// walks one PDS's `listRepos` and hands the repos we don't know yet to the crawler worker.#[derive(Clone)]287 unmodified linesExpand all 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, skew_secs)); } else if is_stuck(db, &repo.did, &stored)? { if let Some(listing) = listing_if_newer(&repo, &stored, skew_secs) { newer.push(listing); } else if known == KnownRepos::ResyncNewerOrStuck && is_stuck(db, &repo.did, &stored)? { stuck.push(repo.did.into_static()); } continue; }40 unmodified lines}
/// `repo` as a listing if it's listed past the rev we holdfn listing_if_newer( repo: ListedRepo<'_>, repo: &ListedRepo<'_>, stored: &RepoState<'_>, skew_secs: i64,) -> Option<RepoListing> { let listing = RepoListing { did: repo.did.into_static(), did: repo.did.clone().into_static(), rev: DbTid::from(&repo.rev), active: repo.active.unwrap_or(true), status: repo.status.into_static(), status: repo.status.clone().into_static(), create_if_missing: false, }; listing.claims_newer(stored, skew_secs).then_some(listing)}183 unmodified linesExpand all