Something went wrong. Try again.
fork of hydrant
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819820821822823824825826827828829830831832833834835836837838839840841842843844845846847848849850851852853854855856857858859860861862863864865866867868869870871872873874875876877878879880881882883884885886887888889890891892893894895896897898899900901902903904905906907908909910911912913914915916917918919920921922923924925926927928929930931932933934935936937938939940941942943944945946947948949950951952953954955956957958959960961962963964965966967968969970971972973974975976977use fjall::Slice;use jacquard_common::IntoStatic;use jacquard_common::types::did::Did;use miette::{IntoDiagnostic, Result};use std::sync::Arc;use tracing::{debug, error, warn};
use crate::backfill::admission::BackfillAdmission;use crate::backfill::client::ThrottledHttpClient;use crate::backfill::error::BackfillError;use crate::config::BackfillStrategy;use crate::db::{Txn as DbTxn, keys};use crate::ingest::indexer::{IndexerMessage, IndexerTx};use crate::state::AppState;use crate::types::{GaugeState, RepoState, RepoStatus, ResyncErrorKind, ResyncState};
use super::process::process_did;
#[derive(Debug, Clone, Copy, PartialEq, Eq)]pub(super) enum TaskDisposition { Finished, Deferred,}
fn stage_excluded_pending_cleanup( txn: &mut DbTxn<'_>, db: &crate::db::Db, did: &Did, pending_key: &[u8],) -> Result<bool> { if !txn.lock_repo_and_is_excluded(did)? { return Ok(false); } txn.transition_pending_key(did, pending_key, GaugeState::Synced)?; txn.batch.remove(&db.indexer.pending, pending_key); Ok(true)}
async fn remove_pending_if_excluded( state: &AppState, did: &Did, pending_key: Slice,) -> Result<bool> { let did = did.clone(); state .db .run(move |db| { let mut txn = DbTxn::new(db); if !stage_excluded_pending_cleanup(&mut txn, db, &did, pending_key.as_ref())? { return Ok(false); } txn.commit()?; Ok(true) }) .await}
async fn remove_pending(state: &AppState, pending_key: Slice) -> Result<()> { state .db .run(move |db| { let mut batch = db.inner.batch(); batch.remove(&db.indexer.pending, pending_key); batch.commit().into_diagnostic() }) .await}
pub(super) async fn did_task( state: &Arc<AppState>, http: ThrottledHttpClient, buffer_tx: IndexerTx, did: &Did, pending_key: Slice, admission: BackfillAdmission, _permit: tokio::sync::OwnedSemaphorePermit, verify_signatures: bool, strategy: BackfillStrategy,) -> Result<TaskDisposition, BackfillError> { let db = &state.db;
// Drop stale/excluded work before resolver, admission, or network I/O. A // repository drop may race task spawning, and orphan pending keys are not // necessarily the one named by current metadata. if remove_pending_if_excluded(state, did, pending_key.clone()).await? { return Ok(TaskDisposition::Finished); }
match process_did( state, &http, did, pending_key.clone(), &admission, verify_signatures, strategy, ) .await { Ok(Some(imported)) => { let applied = state .db .run({ let did = did.clone(); let pending_key = pending_key.clone(); move |db| { let did_key = keys::repo_key(&did); let mut txn = DbTxn::new(db); if stage_excluded_pending_cleanup(&mut txn, db, &did, pending_key.as_ref())? { txn.commit()?; return Ok::<_, miette::Report>(false); } let applied = txn.transition_pending_key( &did, pending_key.as_ref(), GaugeState::Synced, )?; txn.batch.remove(&db.indexer.pending, pending_key.clone()); if applied { txn.batch.remove(&db.indexer.resync, &did_key); } txn.commit()?; Ok::<_, miette::Report>(applied) } }) .await?;
if !applied { return Ok(TaskDisposition::Finished); }
state .backfill_drained .fetch_add(1, std::sync::atomic::Ordering::Relaxed);
let state = state.clone(); state .db .run(move |db| { db.inner .persist(fjall::PersistMode::Buffer) .into_diagnostic() }) .await?;
#[cfg(feature = "indexer_stream")] imported.emit(&state, did); #[cfg(not(feature = "indexer_stream"))] drop(imported);
if let Err(e) = buffer_tx .send(IndexerMessage::BackfillFinished(did.clone())) .await { error!(err = %e, "failed to send BackfillFinished"); } Ok(TaskDisposition::Finished) } Ok(None) => { remove_pending(state, pending_key).await?; Ok(TaskDisposition::Finished) } Err(BackfillError::PdsBusy) => Ok(TaskDisposition::Deferred), Err(BackfillError::UnsafeEndpoint(reason)) => { // The identity document is the only authority that can repair a // permanently unsafe PDS. Park this queue item rather than hot // retrying the same invalid endpoint. A later identity event can // enqueue fresh work with new evidence. warn!(%reason, "unsafe PDS endpoint, parking backfill until identity evidence changes"); remove_pending(state, pending_key).await?; Ok(TaskDisposition::Finished) } Err(BackfillError::Deleted) => { warn!("orphaned pending entry, cleaning up"); state .db .run({ let did = did.clone(); let pending_key = pending_key.clone(); move |db| { let mut txn = DbTxn::new(db); if stage_excluded_pending_cleanup(&mut txn, db, &did, pending_key.as_ref())? { txn.commit()?; return Ok::<_, miette::Report>(()); } txn.transition_pending_key(&did, pending_key.as_ref(), GaugeState::Synced)?; txn.batch.remove(&db.indexer.pending, pending_key); txn.commit()?; Ok::<_, miette::Report>(()) } }) .await?; state .backfill_drained .fetch_add(1, std::sync::atomic::Ordering::Relaxed); Ok(TaskDisposition::Finished) } Err(e) => { match &e { BackfillError::Ratelimited | BackfillError::PreemptivelyThrottled => { debug!("too many requests"); } BackfillError::Transport(reason) => { error!(%reason, "transport error"); } BackfillError::Generic(e) => { error!(err = %e, "failed"); } BackfillError::RepoNotFound => { warn!("repo not found, retrying through existing backoff"); } BackfillError::Deleted => unreachable!("already handled"), BackfillError::PdsBusy => unreachable!("already handled"), BackfillError::UnsafeEndpoint(_) => unreachable!("handled above"), }
let error_kind = match &e { BackfillError::Ratelimited | BackfillError::PreemptivelyThrottled => { ResyncErrorKind::Ratelimited } BackfillError::Transport(_) => ResyncErrorKind::Transport, BackfillError::Generic(_) => ResyncErrorKind::Generic, BackfillError::RepoNotFound => ResyncErrorKind::Generic, BackfillError::Deleted => unreachable!("already handled"), BackfillError::PdsBusy => unreachable!("already handled"), BackfillError::UnsafeEndpoint(_) => unreachable!("handled above"), };
let did_key = keys::repo_key(did);
// 1. get current retry count let did_key_clone = did_key.clone(); let existing_state = db .run(move |db| { db.indexer .resync .get(&did_key_clone) .into_diagnostic() .and_then(|b| { b.map(|b| rmp_serde::from_slice::<ResyncState>(&b).into_diagnostic()) .transpose() }) }) .await?;
let mut retry_count = match existing_state { Some(ResyncState::Error { retry_count, .. }) => retry_count, Some(ResyncState::Gone { .. }) => { remove_pending(state, pending_key).await?; return Ok(TaskDisposition::Finished); } None => 0, };
// Calculate new stats let next_retry = if matches!(e, BackfillError::PreemptivelyThrottled) { chrono::Utc::now().timestamp() + 60 } else { retry_count += 1; ResyncState::next_backoff(retry_count) };
let resync_state = ResyncState::Error { kind: error_kind, retry_count, next_retry, }; let error_string = e.to_string();
let applied = state .db .run({ let did_key = did_key.into_static(); let did = did.clone(); let pending_key = pending_key.clone(); move |db| { // 3. save to resync let serialized_resync_state = rmp_serde::to_vec(&resync_state).into_diagnostic()?;
// 4. and update the main repo state let serialized_repo_state = if let Some(state_bytes) = db.repos.get(&did_key).into_diagnostic()? { let mut state: RepoState = rmp_serde::from_slice(&state_bytes).into_diagnostic()?; state.active = true; state.status = RepoStatus::Error(error_string.into()); Some(rmp_serde::to_vec(&state).into_diagnostic()?) } else { None }; let mut txn = DbTxn::new(db); if stage_excluded_pending_cleanup(&mut txn, db, &did, pending_key.as_ref())? { txn.commit()?; return Ok::<_, miette::Report>(false); } let applied = txn.transition_pending_key( &did, pending_key.as_ref(), GaugeState::Resync(Some(error_kind)), )?; txn.batch.remove(&db.indexer.pending, pending_key.clone()); if applied { txn.batch .insert(&db.indexer.resync, &did_key, serialized_resync_state); if let Some(state_bytes) = serialized_repo_state { txn.batch.insert(&db.repos, &did_key, state_bytes); } } txn.commit()?; Ok::<_, miette::Report>(applied) } }) .await?;
if applied { state .backfill_drained .fetch_add(1, std::sync::atomic::Ordering::Relaxed); }
Err(e) } }}
#[cfg(test)]mod tests { use super::*; use crate::backfill::admission::BackfillAdmission; use crate::config::Config; use crate::db; use crate::db::types::DbRkey; use crate::ingest::indexer::FirehoseWorker; use crate::types::RepoMetadata; use axum::Router; use axum::routing::get; use bytes::Bytes; use jacquard_common::CowStr; use jacquard_common::types::cid::IpldCid; use jacquard_common::types::string::Handle; use jacquard_common::types::tid::Tid; use jacquard_repo::commit::Commit as AtpCommit; use jacquard_repo::mst::Mst; use jacquard_repo::{BlockStore, MemoryBlockStore}; use reqwest::StatusCode; use std::sync::atomic::{AtomicUsize, Ordering}; use tokio::sync::{RwLock, Semaphore}; use url::Url;
#[tokio::test] async fn excluded_orphan_pending_entry_is_removed_before_work() -> miette::Result<()> { let tmp = tempfile::tempdir().into_diagnostic()?; let config = Config { database_path: tmp.path().to_path_buf(), ..Default::default() }; let state = AppState::new(&config)?; let did = Did::new("did:plc:ewvi7nxzyoun6zhxrhs64oiz")?.into_static(); let pending_key = keys::pending_key(7); let mut batch = state.db.inner.batch(); batch.insert( &state.db.filter, crate::db::filter::exclude_key(did.as_str())?, [], ); batch.insert(&state.db.indexer.pending, pending_key, keys::repo_key(&did)); batch.commit().into_diagnostic()?;
assert!( remove_pending_if_excluded(&state, &did, fjall::Slice::from(pending_key.as_slice()),) .await? ); assert!( state .db .indexer .pending .get(pending_key) .into_diagnostic()? .is_none() ); Ok(()) }
// --- full-backfill `RepoNotFound` boundary --------------------------------------------- // // `getRepo` `RepoNotFound` is a fetch result, not lifecycle evidence. these tests pin the // non-destructive policy at the task boundary: the snapshot survives, the queue parks a // retryable error, and only a later valid CAR or an authoritative deletion moves state.
const DID: &str = "did:plc:ewvi7nxzyoun6zhxrhs64oiz"; const NOT_FOUND_BODY: &[u8] = br#"{"error":"RepoNotFound","message":"not here"}"#; const COLLECTION: &str = "app.bsky.feed.post";
fn post(text: &str) -> Vec<u8> { serde_ipld_dagcbor::to_vec(&serde_json::json!({ "$type": COLLECTION, "text": text, })) .unwrap() }
struct PdsStub { base: Url, address: std::net::SocketAddr, response: Arc<RwLock<(StatusCode, Bytes)>>, requests: Arc<AtomicUsize>, }
impl PdsStub { async fn spawn() -> Self { let response = Arc::new(RwLock::new(( StatusCode::BAD_REQUEST, Bytes::from_static(NOT_FOUND_BODY), ))); let requests = Arc::new(AtomicUsize::new(0)); let app = Router::new().route( "/xrpc/com.atproto.sync.getRepo", get({ let response = response.clone(); let requests = requests.clone(); move || { let response = response.clone(); let requests = requests.clone(); async move { requests.fetch_add(1, Ordering::SeqCst); let (status, body) = response.read().await.clone(); (status, body) } } }), ); let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); let address = listener.local_addr().unwrap(); // Keep the endpoint public by syntax while pinning its socket in // the test client; no DNS or external network is involved. let base = Url::parse("http://pds.example/").unwrap(); tokio::spawn(async move { axum::serve(listener, app).await.unwrap() }); Self { base, address, response, requests, } }
async fn serve_car(&self, car: Bytes) { *self.response.write().await = (StatusCode::OK, car); }
fn requests(&self) -> usize { self.requests.load(Ordering::SeqCst) } }
async fn spawn_doc_stub(doc: String) -> Url { let app = Router::new().fallback({ move |_uri: axum::http::Uri| { let doc = doc.clone(); async move { ( [(axum::http::header::CONTENT_TYPE, "application/json")], doc, ) } } }); let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); let base = Url::parse(&format!("http://{}/", listener.local_addr().unwrap())).unwrap(); tokio::spawn(async move { axum::serve(listener, app).await.unwrap() }); base }
fn did_doc(base: &Url) -> String { serde_json::json!({ "@context": ["https://www.w3.org/ns/did/v1"], "id": DID, "service": [{ "id": "#atproto_pds", "type": "AtprotoPersonalDataServer", "serviceEndpoint": base.as_str(), }], }) .to_string() }
fn rooted(rev: &str) -> miette::Result<RepoState<'static>> { let mut state = RepoState::synced(); state.root = Some(crate::types::Commit { version: 3, rev: crate::db::types::DbTid::from(&Tid::new(rev).into_diagnostic()?), data: jacquard_repo::mst::util::compute_cid(b"seeded-root").into_diagnostic()?, prev: None, sig: Bytes::from_static(b"signature"), }); state.pds = Some(CowStr::Borrowed("https://old-pds.example")); state.handle = Some(Handle::new_static("alice.test").into_diagnostic()?); state.last_message_time = Some(101); state.last_updated_at = 5; Ok(state.into_static()) }
struct BuiltCar { bytes: Bytes, mst_root: IpldCid, rev: Tid, }
async fn build_car(rev: &str, records: &[(&str, &[u8])]) -> miette::Result<BuiltCar> { let store = Arc::new(MemoryBlockStore::new()); let mut mst = Mst::new(store.clone()); for &(rkey, body) in records { let cid = jacquard_repo::mst::util::compute_cid(body).into_diagnostic()?; mst.add_mut(&format!("{COLLECTION}/{rkey}"), cid) .await .into_diagnostic()?; store .put_many([(cid, Bytes::copy_from_slice(body))]) .await .into_diagnostic()?; } let mst_root = mst.persist().await.into_diagnostic()?; let rev = Tid::new(rev).into_diagnostic()?; let did: Did = Did::new_static(DID).into_diagnostic()?; let commit = AtpCommit { did, version: 3, data: mst_root, rev: rev.clone(), prev: None, sig: Bytes::from_static(b"signature"), }; let commit_cid = commit.to_cid().into_diagnostic()?; let commit_cbor = commit.to_cbor().into_diagnostic()?; store .put_many([(commit_cid, Bytes::from(commit_cbor.clone()))]) .await .into_diagnostic()?;
let mut bytes = Vec::new(); let mut writer = iroh_car::CarWriter::new(iroh_car::CarHeader::new_v1(vec![commit_cid]), &mut bytes); writer .write(commit_cid, &commit_cbor) .await .into_diagnostic()?; mst.write_blocks_to_car(&mut writer) .await .into_diagnostic()?; writer.finish().await.into_diagnostic()?; Ok(BuiltCar { bytes: Bytes::from(bytes), mst_root, rev, }) }
struct Fixture { state: Arc<AppState>, did: Did, pds: PdsStub, _tmp: tempfile::TempDir, }
impl Fixture { async fn new() -> miette::Result<Self> { let pds = PdsStub::spawn().await; let endpoint = pds.base.clone(); Self::new_with_pds(pds, endpoint).await }
async fn new_with_pds(pds: PdsStub, pds_endpoint: Url) -> miette::Result<Self> { let tmp = tempfile::tempdir().into_diagnostic()?; let doc_base = spawn_doc_stub(did_doc(&pds_endpoint)).await; let public_plc = Url::parse("http://public.example/").unwrap(); let config = Config { database_path: tmp.path().to_path_buf(), plc_urls: vec![public_plc.clone()], identity_cache_size: 64, ..Default::default() }; let mut app_state = AppState::new(&config)?; let public_client = reqwest::Client::builder() .no_proxy() .resolve( "public.example", std::net::SocketAddr::from(( std::net::Ipv4Addr::LOCALHOST, doc_base.port().unwrap(), )), ) .build() .into_diagnostic()?; let resolver = crate::resolver::Resolver::with_plc_urls_and_client( vec![public_plc.clone()], 64, crate::net::PublicHttpClient::test(public_client), ); app_state.resolver = resolver; Ok(Self { state: Arc::new(app_state), did: Did::new_static(DID).into_diagnostic()?, pds, _tmp: tmp, }) }
fn seed( &self, repo: &RepoState<'static>, index_id: u64, records: &[(&str, &[u8])], ) -> miette::Result<()> { let mut batch = self.state.db.inner.batch(); batch.insert( &self.state.db.repos, keys::repo_key(&self.did), db::ser_repo_state(repo)?, ); batch.insert( &self.state.db.repo_metadata, keys::repo_metadata_key(&self.did), db::ser_repo_meta(&RepoMetadata::backfilling(index_id))?, ); batch.insert( &self.state.db.indexer.pending, keys::pending_key(index_id), keys::repo_key(&self.did), ); for &(rkey, body) in records { self.state.db.indexer.stage_record( &mut batch, keys::record_key(&self.did, COLLECTION, &DbRkey::new(rkey)), body, ); } batch.commit().into_diagnostic() }
/// mirrors the retry worker: a completed queue entry gives way to a fresh pending key. fn reindex_pending(&self, from: u64, to: u64) -> miette::Result<()> { let mut batch = self.state.db.inner.batch(); batch.remove(&self.state.db.indexer.pending, keys::pending_key(from)); batch.insert( &self.state.db.indexer.pending, keys::pending_key(to), keys::repo_key(&self.did), ); batch.insert( &self.state.db.repo_metadata, keys::repo_metadata_key(&self.did), db::ser_repo_meta(&RepoMetadata::backfilling(to))?, ); batch.commit().into_diagnostic() }
fn clear_resync(&self) -> miette::Result<()> { self.state .db .indexer .resync .remove(keys::repo_key(&self.did)) .into_diagnostic()?; Ok(()) }
/// the authoritative deletion primitive plus its `Deleted` tombstone: the same pair the /// firehose and list-repos paths apply under the repository lock. fn delete_authoritatively(&self, repo: &RepoState<'static>) -> miette::Result<()> { let mut txn = DbTxn::new(&self.state.db); crate::ops::delete_repo(&mut txn, &self.state.db, &self.did, repo)?; crate::ops::transition_repo( &mut txn.batch, &self.state.db, &mut Vec::new(), &self.did, repo.clone(), RepoStatus::Deleted, )?; txn.commit() }
async fn run(&self, pending_key: [u8; 8]) -> Result<TaskDisposition, BackfillError> { let (tx, _worker) = FirehoseWorker::new(self.state.clone(), 1); let client = reqwest::Client::builder() .resolve("pds.example", self.pds.address) .build() .into_diagnostic()?; let http = ThrottledHttpClient::new( crate::net::PublicHttpClient::test(client), self.state.throttler.clone(), ); let admission = BackfillAdmission::new(4, 4); let permit = Arc::new(Semaphore::new(1)).acquire_owned().await.unwrap(); did_task( &self.state, http, tx, &self.did, Slice::from(pending_key), admission, permit, false, BackfillStrategy::Full, ) .await }
fn repo(&self) -> miette::Result<RepoState<'static>> { let bytes = self .state .db .repos .get(keys::repo_key(&self.did)) .into_diagnostic()? .expect("repository state row") .to_vec(); Ok(db::deser_repo_state(&bytes)?.into_static()) }
fn resync(&self) -> miette::Result<Option<ResyncState>> { self.state .db .indexer .resync .get(keys::repo_key(&self.did)) .into_diagnostic()? .map(|bytes| rmp_serde::from_slice::<ResyncState>(&bytes).into_diagnostic()) .transpose() }
fn pending_keys(&self) -> miette::Result<Vec<Vec<u8>>> { self.state .db .indexer .pending .iter() .map(|guard| Ok(guard.into_inner().into_diagnostic()?.0.to_vec())) .collect() }
fn record_rkeys(&self) -> miette::Result<Vec<String>> { let prefix = keys::record_prefix_did(&self.did); self.state .db .indexer .record_prefix(&prefix) .map(|guard| { let (key, _) = guard.into_inner().into_diagnostic()?; let (_, rkey) = keys::split_record_suffix(&key[prefix.len()..])?; Ok(rkey.to_string()) }) .collect() } }
#[tokio::test] async fn unsafe_pds_endpoint_is_parked_until_fresh_identity_evidence() -> miette::Result<()> { let pds = PdsStub::spawn().await; let fixture = Fixture::new_with_pds(pds, Url::parse("http://127.0.0.1/").unwrap()).await?; fixture.seed(&RepoState::backfilling().into_static(), 13, &[])?;
assert_eq!( fixture.run(keys::pending_key(13)).await?, TaskDisposition::Finished, "unsafe endpoint work must park instead of entering retry backoff" ); assert!(fixture.pending_keys()?.is_empty()); assert!(fixture.resync()?.is_none()); assert_eq!( fixture.pds.requests(), 0, "unsafe endpoint must not connect" ); Ok(()) }
#[tokio::test] async fn repo_not_found_keeps_a_rooted_snapshot_and_parks_a_retry() -> miette::Result<()> { let fixture = Fixture::new().await?; let seeded = rooted("3jzfcijpj2z2a")?; let body = post("a"); fixture.seed(&seeded, 7, &[("a", body.as_slice())])?;
let err = fixture.run(keys::pending_key(7)).await.unwrap_err();
assert!( matches!(err, BackfillError::RepoNotFound), "unexpected error: {err}" ); let after = fixture.repo()?; assert_eq!( after.root.as_ref().map(|root| root.data), seeded.root.as_ref().map(|root| root.data), "a non-authoritative fetch result must not clear the root" ); assert!( matches!(&after.status, RepoStatus::Error(_)), "expected a retryable error status, got {:?}", after.status ); assert!(after.active); assert_eq!(after.last_message_time, Some(101)); assert_eq!(fixture.record_rkeys()?, ["a"]); assert!( fixture .state .db .repo_metadata .get(keys::repo_metadata_key(&fixture.did)) .into_diagnostic()? .is_some(), "metadata must survive" ); assert!( fixture.pending_keys()?.is_empty(), "parked work must leave pending, or the worker spins" );
let Some(ResyncState::Error { kind, retry_count, next_retry, }) = fixture.resync()? else { panic!("expected a durable retryable error"); }; assert_eq!(kind, ResyncErrorKind::Generic); assert_eq!(retry_count, 1); assert!( next_retry > chrono::Utc::now().timestamp(), "the retry worker may only requeue a due error" ); Ok(()) }
#[tokio::test] async fn repo_not_found_on_a_never_rooted_repo_uses_the_same_retry_policy() -> miette::Result<()> { let fixture = Fixture::new().await?; fixture.seed(&RepoState::backfilling().into_static(), 9, &[])?;
let err = fixture.run(keys::pending_key(9)).await.unwrap_err();
assert!( matches!(err, BackfillError::RepoNotFound), "unexpected error: {err}" ); let after = fixture.repo()?; assert!(after.root.is_none(), "no root may be invented"); assert!( matches!(&after.status, RepoStatus::Error(_)), "expected a retryable error status, got {:?}", after.status ); assert!(fixture.record_rkeys()?.is_empty()); assert!(fixture.pending_keys()?.is_empty()); assert!(matches!( fixture.resync()?, Some(ResyncState::Error { kind: ResyncErrorKind::Generic, .. }) )); Ok(()) }
#[tokio::test] async fn repo_not_found_cannot_overwrite_a_replaced_pending_entry() -> miette::Result<()> { let fixture = Fixture::new().await?; let seeded = rooted("3jzfcijpj2z2a")?; let body = post("a"); fixture.seed(&seeded, 7, &[("a", body.as_slice())])?; fixture.reindex_pending(7, 11)?;
let err = fixture.run(keys::pending_key(7)).await.unwrap_err();
assert!( matches!(err, BackfillError::RepoNotFound), "unexpected error: {err}" ); assert_eq!(fixture.pending_keys()?, [keys::pending_key(11).to_vec()]); assert!( fixture.resync()?.is_none(), "stale work must not park a retry against a newer queue entry" ); let after = fixture.repo()?; assert_eq!(after.status, RepoStatus::Synced); assert_eq!( after.root.as_ref().map(|root| root.data), seeded.root.as_ref().map(|root| root.data) ); assert_eq!(fixture.record_rkeys()?, ["a"]); Ok(()) }
#[tokio::test] async fn authoritative_deletion_is_not_resurrected_by_a_late_repo_not_found() -> miette::Result<()> { let fixture = Fixture::new().await?; let seeded = rooted("3jzfcijpj2z2a")?; let body = post("a"); fixture.seed(&seeded, 7, &[("a", body.as_slice())])?; fixture.delete_authoritatively(&seeded)?;
let err = fixture.run(keys::pending_key(7)).await.unwrap_err();
assert!( matches!(err, BackfillError::RepoNotFound), "unexpected error: {err}" ); let after = fixture.repo()?; assert_eq!(after.status, RepoStatus::Deleted); assert!(!after.active, "the tombstone must stay inactive"); assert!( fixture.record_rkeys()?.is_empty(), "erased records must stay erased" ); assert!(fixture.pending_keys()?.is_empty()); assert!( fixture.resync()?.is_none(), "a deleted repo must not be re-queued" ); Ok(()) }
#[tokio::test] async fn a_later_valid_car_replaces_the_error_after_requeue() -> miette::Result<()> { let fixture = Fixture::new().await?; let seeded = rooted("3jzfcijpj2z2a")?; let first = post("a"); fixture.seed(&seeded, 7, &[("a", first.as_slice())])?; assert!(matches!( fixture.run(keys::pending_key(7)).await.unwrap_err(), BackfillError::RepoNotFound )); assert_eq!(fixture.pds.requests(), 1);
fixture.clear_resync()?; fixture.reindex_pending(7, 21)?; let second = post("b"); let car = build_car( "3jzfcijpj2z2b", &[("a", first.as_slice()), ("b", second.as_slice())], ) .await?; fixture.pds.serve_car(car.bytes.clone()).await;
assert_eq!( fixture.run(keys::pending_key(21)).await.unwrap(), TaskDisposition::Finished );
let after = fixture.repo()?; assert_eq!( after.root.as_ref().map(|root| root.data), Some(car.mst_root) ); assert_eq!( after.root.as_ref().map(|root| root.rev.to_tid()), Some(car.rev.clone()) ); assert_eq!(fixture.record_rkeys()?, ["a", "b"]); assert!( fixture.resync()?.is_none(), "a successful import clears the error row" ); assert!(fixture.pending_keys()?.is_empty()); assert_eq!(fixture.pds.requests(), 2); Ok(()) }}