Something went wrong. Try again.
fork of hydrant
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764use std::collections::HashMap;use std::sync::Arc;use std::time::Instant;
use bytes::Bytes;use fjall::Slice;use miette::{IntoDiagnostic, Result};use reqwest::StatusCode;use smol_str::{SmolStr, ToSmolStr};use tracing::{debug, trace, warn};
use jacquard_api::com_atproto::sync::get_repo::GetRepoError;use jacquard_common::IntoStatic;use jacquard_common::types::cid::Cid as AtCid;use jacquard_common::types::did::Did;use jacquard_repo::mst::Mst;use jacquard_repo::{BlockStore, MemoryBlockStore};
use crate::backfill::admission::BackfillAdmission;use crate::backfill::client::{ThrottledHttpClient, collect_body_bounded};use crate::backfill::error::BackfillError;use crate::backfill::sparse::{SparseBackfillResult, process_did_sparse};use crate::config::{BackfillStrategy, RateTier};use crate::db::types::{DbAction, DbRkey};use crate::db::{self, Txn as DbTxn, keys};use crate::filter::FilterMode;use crate::net::PublicEndpoint;use crate::ops;use crate::sparse_mst::sparse_probe_collection;use crate::state::AppState;use crate::types::{Commit, GaugeState, RepoState, RepoStatus, ResyncState};use crate::util::{parse_retry_after, throttle::ThrottleHandle};
#[cfg(feature = "indexer_stream")]use crate::types::{AccountEvt, BroadcastEvent};#[cfg(feature = "indexer_stream")]use std::sync::atomic::Ordering;
/// without the stream feature there is nobody to announce the transition to,/// so both states are carried and dropped.#[cfg_attr(not(feature = "indexer_stream"), allow(dead_code))]pub(super) struct BackfillImported { before: RepoState<'static>, after: RepoState<'static>,}
impl BackfillImported { #[cfg(feature = "indexer_stream")] pub(super) fn emit(self, app_state: &AppState, did: &Did) { if self.after.account_moved_from(&self.before) { emit_account(app_state, did, &self.after.status, self.after.active); } if self.after.identity_moved_from(&self.before) { ops::announce_identity(&app_state.db, did, self.after.handle); } }}
#[cfg(feature = "indexer_stream")]fn emit_account(app_state: &AppState, did: &Did, status: &RepoStatus, active: bool) { let status = match status { RepoStatus::Deactivated => "deactivated", RepoStatus::Takendown => "takendown", RepoStatus::Suspended => "suspended", RepoStatus::Deleted => "deleted", RepoStatus::Desynchronized => "desynchronized", RepoStatus::Throttled => "throttled", _ => "", }; let evt = AccountEvt { did: did.clone(), active, status: (!status.is_empty()).then(|| status.into()), }; let _ = app_state .db .stream .event_tx .send(ops::make_account_event(&app_state.db, evt));}
pub(super) async fn process_did( app_state: &Arc<AppState>, http: &ThrottledHttpClient, did: &Did, pending_key: Slice, admission: &BackfillAdmission, verify_signatures: bool, strategy: BackfillStrategy,) -> Result<Option<BackfillImported>, BackfillError> { debug!("starting...");
// always invalidate doc before backfilling app_state.resolver.invalidate(did).await;
let db = &app_state.db; let did_key = keys::repo_key(did); let Some(state_bytes) = db .run({ let did_key = did_key.clone(); move |db| db.repos.get(did_key).into_diagnostic() }) .await? else { return Err(BackfillError::Deleted); }; let mut state: RepoState<'static> = rmp_serde::from_slice::<RepoState>(&state_bytes) .into_diagnostic()? .into_static(); let previous_state = state.clone();
// 1. resolve pds let start = Instant::now(); let doc = app_state.resolver.resolve_doc(did).await?; let pds = doc.pds.clone(); if let Err(error) = PublicEndpoint::parse_http(&pds) { return Err(BackfillError::UnsafeEndpoint( format!("unsafe public PDS endpoint {pds}: {error}").into(), )); } trace!( pds = %doc.pds, handle = %doc.handle.as_deref().unwrap_or("handle.invalid"), elapsed = %start.elapsed().as_secs_f32(), "resolved to pds" ); state.update_from_doc(doc);
let Some(admission_permit) = admission.try_acquire(&pds).await else { return Err(BackfillError::PdsBusy); };
if strategy != BackfillStrategy::Full { let filter = app_state.filter.load(); let sparse_supported = !filter.collections.is_empty() && sparse_probe_collection(&filter.collections).is_some(); if sparse_supported { match process_did_sparse( app_state, http, did, &pds, state.clone(), verify_signatures, strategy, ) .await { Ok(SparseBackfillResult::Imported(sparse)) => { trace!( records = sparse.records, node_blocks = sparse.node_blocks, node_bytes = sparse.node_bytes, active = sparse.state.active, status = ?sparse.state.status, "sparse backfill complete" ); return Ok(Some(BackfillImported { before: previous_state, after: sparse.state, })); } Ok(SparseBackfillResult::Discarded) => { remove_discarded_repo(app_state, did, &pending_key).await?; return Ok(None); } Ok(SparseBackfillResult::Skipped) => { debug!("sparse backfill skipped, falling back to full getRepo"); } Err( e @ (BackfillError::Ratelimited | BackfillError::PreemptivelyThrottled | BackfillError::Transport(_)), ) => { return Err(e); } Err(e) => { warn!(err = %e, "sparse backfill failed, falling back to full getRepo"); } } } else if strategy == BackfillStrategy::SparseFilter { debug!("sparse backfill requested but filter is not sparse-compatible"); } }
// 2. fetch repo (car) let start = Instant::now(); let throttle = app_state.throttler.get_handle(&pds).await; let tier = app_state.resolve_pds_tier(pds.host_str().unwrap_or("")); let car_bytes = match fetch_full_repo_car( http, &pds, did, &throttle, &tier, app_state.max_car_body_bytes, ) .await? { FullRepoOutcome::Car(body) => body, FullRepoOutcome::NotFound => { // `RepoNotFound` is not lifecycle evidence: it says the pds has no repo, not that the // account was deleted, and the repo may only be missing on this target. erasure stays // reserved for authoritative firehose and listing `Deleted` events, so fail retryably // and let the task error path schedule the existing jittered retry. warn!("repo not found on pds, retrying"); return Err(BackfillError::RepoNotFound); } FullRepoOutcome::Inactive(status) => { warn!(?status, "repo is inactive, stopping backfill");
#[cfg(feature = "indexer_stream")] let event_status = status.clone(); let resync_state = ResyncState::Gone { status: status.clone(), }; let resync_bytes = rmp_serde::to_vec(&resync_state).into_diagnostic()?;
let app_state_clone = app_state.clone(); #[cfg(feature = "indexer_stream")] let event_did = did.clone(); let did = did.clone(); let pending_key = pending_key.clone(); let applied = app_state_clone .db .run(move |db| { let mut txn = DbTxn::new(db); if txn.lock_repo_and_is_excluded(&did)? { return Ok::<_, miette::Report>(false); } let applied = txn.transition_pending_key( &did, pending_key.as_ref(), GaugeState::Resync(None), )?; txn.batch.remove(&db.indexer.pending, pending_key.clone()); if applied { crate::db::Db::update_repo_state( &mut txn.batch, &db.repos, &did, move |state, (key, batch)| { state.active = false; state.status = status; batch.insert(&db.indexer.resync, key, resync_bytes); Ok((true, ())) }, )?; } txn.commit()?; Ok::<_, miette::Report>(applied) }) .await?;
#[cfg(feature = "indexer_stream")] if applied { emit_account(app_state, &event_did, &event_status, false); } #[cfg(not(feature = "indexer_stream"))] let _ = applied;
return Ok(None); } }; drop(admission_permit);
trace!( bytes = car_bytes.len(), elapsed = ?start.elapsed(), "fetched car bytes" );
// 3. import repo let start = Instant::now(); let parsed = crate::car::parse_car(car_bytes)?; trace!(elapsed = %start.elapsed().as_secs_f32(), "parsed car");
let start = Instant::now(); let root_cid = parsed.root; if app_state.verify_cids { crate::car::validate_block_cids(&parsed.blocks)?; } let store = Arc::new(MemoryBlockStore::new_from_blocks(parsed.blocks)); trace!( blocks = store.len(), elapsed = ?start.elapsed(), "stored blocks in memory" );
// 4. parse root commit to get mst root let root_bytes = store .get(&root_cid) .await .into_diagnostic()? .ok_or_else(|| miette::miette!("root block missing from CAR"))?;
let root_commit = jacquard_repo::commit::Commit::<jacquard_common::DefaultStr>::from_cbor(&root_bytes) .into_diagnostic()?; debug!( rev = %root_commit.rev, cid = %root_commit.data, "repo at revision" );
// 4.5. verify commit signature if verify_signatures { let pubkey = app_state.resolver.resolve_signing_key(did).await?; root_commit .verify(&pubkey) .map_err(|e| miette::miette!("signature verification failed for {did}: {e}"))?; trace!("signature verified"); }
let root_commit = Commit::from(root_commit);
// 5. walk mst and fetch every record block under one store lock let start = Instant::now(); let mst: Mst<MemoryBlockStore> = Mst::load(store, root_commit.data, None); let leaves = mst.leaves().await.into_diagnostic()?; let leaf_cids = leaves.iter().map(|(_, cid)| *cid).collect::<Vec<_>>(); let leaf_blocks = mst.storage().get_many(&leaf_cids).await.into_diagnostic()?; let records = leaves.into_iter().zip(leaf_blocks); drop(mst); trace!(elapsed = %start.elapsed().as_secs_f32(), "walked mst");
// 6. insert records into db let start = Instant::now(); let result = { let app_state = app_state.clone(); let did = did.clone(); let rev = root_commit.rev;
tokio::task::spawn_blocking(move || { let filter = app_state.filter.load(); let mut count = 0; let mut collection_counts: HashMap<SmolStr, u64> = HashMap::new(); let mut txn = DbTxn::new(&app_state.db); let mut record_txn = txn.backfill_records(&app_state, &rev, &did)?; if record_txn.is_excluded() { return Ok::<_, miette::Report>(None); } // the MST and non-record blocks were released before entering this blocking // persistence phase; records contains only leaf metadata and payload slices.
let prefix = keys::record_prefix_did(&did); let mut existing_cids: HashMap<(SmolStr, DbRkey), SmolStr> = HashMap::new();
if !app_state.ephemeral { for guard in app_state.db.indexer.record_prefix(&prefix) { let (key, value) = guard.into_inner().into_diagnostic()?; let (collection, rkey) = keys::split_record_suffix(&key[prefix.len()..]) .map_err(|e| { miette::miette!("invalid record key '{key:?}' for {did}: {e}") })?;
// the cid is no longer stored; recompute it from the body // (or parse it in links-only mode, which stores cids) let cid = crate::db::cid_from_record_value(&value) .map_err(|e| miette::miette!("invalid record value for {did}: {e}"))?;
existing_cids.insert((collection.into(), rkey), cid); } }
let mut signal_seen = filter.mode == FilterMode::Full || filter.signals.is_empty();
for ((key, cid), val) in records.filter_map(|(leaf, value)| value.map(|value| (leaf, value))) { let (collection, rkey) = ops::parse_path(&key)?;
if !filter.matches_collection(collection) { continue; }
if !signal_seen && filter.matches_signal(collection) { debug!(collection = %collection, "signal matched"); signal_seen = true; }
let rkey = DbRkey::new(rkey); let path = (collection.to_smolstr(), rkey.clone()); let cid_obj: AtCid = AtCid::ipld(cid);
*collection_counts.entry(path.0.clone()).or_default() += 1;
let action = existing_cids.remove(&path).map_or( Some(DbAction::Create), |existing_cid| { (existing_cid != cid_obj.as_str()).then_some(DbAction::Update) }, ); let Some(action) = action else { trace!(collection = %collection, rkey = %rkey, cid = %cid, "skip unchanged record"); continue; }; trace!(collection = %collection, rkey = %rkey, cid = %cid, ?action, "action record");
record_txn.put_record(collection, &rkey, &cid, &val, action)?;
count += 1; }
// remove any remaining existing records (they weren't in the new MST) for ((collection, rkey), cid) in existing_cids { trace!(collection = %collection, rkey = %rkey, cid = %cid, "remove existing record");
record_txn.delete_record(&collection, &rkey)?; count += 1; }
if !signal_seen { trace!(signals = ?filter.signals, "no signal-matching records found, discarding repo"); return Ok::<_, miette::Report>(None); } state.root = Some(root_commit); state.touch(); record_txn.update_repo_state(&state)?; record_txn.finish()?;
let metadata_key = keys::repo_metadata_key(&did); let metadata_bytes = app_state .db .repo_metadata .get(&metadata_key) .into_diagnostic()? .ok_or_else(|| miette::miette!("repo metadata not found for {}", did))?; let mut metadata = crate::db::deser_repo_meta(&metadata_bytes)?; metadata.tracked = true; txn.batch.insert( &app_state.db.repo_metadata, &metadata_key, crate::db::ser_repo_meta(&metadata)?, );
// add the counts db::replace_record_counts( &mut txn.batch, &app_state.db, &did, collection_counts.iter().map(|(col, cnt)| (col.as_str(), *cnt)), )?;
txn.commit()?;
Ok::<_, miette::Report>(Some((count, state))) }) .await .into_diagnostic()?? };
let Some((count, state)) = result else { // signal mode: no signal-matching records found, clean up the optimistically-added repo remove_discarded_repo(app_state, did, &pending_key).await?; return Ok(None); };
trace!(ops = count, elapsed = %start.elapsed().as_secs_f32(), "did ops"); trace!( elapsed = %start.elapsed().as_secs_f32(), "committed backfill batch" );
#[cfg(feature = "indexer_stream")] let _ = db.stream.event_tx.send(BroadcastEvent::Persisted( db.stream.next_event_id.load(Ordering::SeqCst) - 1, ));
trace!("complete"); Ok(Some(BackfillImported { before: previous_state, after: state, }))}
async fn remove_discarded_repo( app_state: &Arc<AppState>, did: &Did, pending_key: &Slice,) -> Result<(), BackfillError> { let did_key = keys::repo_key(did); let metadata_key = keys::repo_metadata_key(did); let did = did.clone(); let pending_key = pending_key.clone(); app_state .db .run(move |db| { let mut txn = DbTxn::new(db); if txn.lock_repo_and_is_excluded(&did)? { return Ok(()); } let applied = txn.transition_pending_key(&did, pending_key.as_ref(), GaugeState::Synced)?; txn.batch.remove(&db.indexer.pending, pending_key); if applied { txn.batch.remove(&db.repos, &did_key); txn.batch.remove(&db.repo_metadata, &metadata_key); txn.counts.add_repos(-1); } txn.commit() }) .await?; Ok(())}
/// upper bound on a buffered xrpc error body. error responses are tiny json objects; this only/// guards against a peer streaming an unbounded body on the error path.const ERROR_BODY_MAX_BYTES: usize = 64 * 1024;
/// outcome of a full-repository `getRepo` fetch that the caller must act on. transport and/// rate-limit failures are surfaced as [`BackfillError`] instead.#[derive(Debug)]enum FullRepoOutcome { /// the streamed CAR body, within the configured size ceiling. Car(Bytes), /// the PDS has no repository for this DID (`RepoNotFound`). this is a fetch result, not /// lifecycle evidence: callers must not delete the stored snapshot for it. NotFound, /// the PDS reported the repository is inactive; carries the status to record. Inactive(RepoStatus),}
/// fetches a full repository CAR by streaming `com.atproto.sync.getRepo` directly through the/// backfill client, enforcing `max_body_bytes` as chunks arrive so a runaway or malicious/// response cannot exhaust memory. this bypasses the typed xrpc layer (which materializes the/// entire decompressed body as one `Vec<u8>` before the caller sees it) while preserving the/// same handling: `RepoNotFound`/`RepoDeactivated`/`RepoTakendown`/`RepoSuspended` are decoded/// from the xrpc error body, and 429 responses feed the throttler from the rate-limit headers.async fn fetch_full_repo_car( http: &ThrottledHttpClient, pds: &url::Url, did: &Did, throttle: &ThrottleHandle, tier: &RateTier, max_body_bytes: usize,) -> Result<FullRepoOutcome, BackfillError> { let pds_endpoint = PublicEndpoint::parse_http(pds).map_err(|error| { BackfillError::Generic(miette::miette!("unsafe public PDS endpoint {pds}: {error}")) })?; let mut url = pds_endpoint .url() .join("xrpc/com.atproto.sync.getRepo") .map_err(|e| BackfillError::Generic(miette::miette!("invalid PDS URL: {e}")))?; url.query_pairs_mut().append_pair("did", did.as_str()); let endpoint = PublicEndpoint::parse_http(&url).map_err(|error| { BackfillError::Generic(miette::miette!( "unsafe public getRepo endpoint {url}: {error}" )) })?;
// hold the per-pds permit for the entire download, not just until response headers: full // repo bodies are large, so releasing early would let unbounded concurrent downloads run // against a single pds and defeat per-pds concurrency (the original xrpc getRepo held it // for the whole download, since its transport read the full body before returning). let _permit = throttle.acquire().await; if throttle.is_throttled() { return Err(BackfillError::PreemptivelyThrottled); } throttle.wait_for_allow(1, tier).await; if throttle.is_throttled() { return Err(BackfillError::PreemptivelyThrottled); } let request = http .get(&endpoint) .header(reqwest::header::ACCEPT, "application/vnd.ipld.car") .build() .map_err(|error| BackfillError::Transport(error.to_string().into()))?; let resp = match http.send_car(request).await { Ok(resp) => { if !throttle.is_throttled() { throttle.record_success(); } resp } Err(e) => { let reason = e.to_string(); if let Some(secs) = throttle.record_failure_detail("transport", reason.clone()) { warn!(%pds, reason, "PDS offline, blacklisting for {secs}s"); } return Err(BackfillError::Transport(reason.into())); } };
let status = resp.status(); if status.is_success() { return match collect_body_bounded(resp, max_body_bytes) .await .map_err(|e| BackfillError::Transport(e.to_string().into()))? { Some(body) => Ok(FullRepoOutcome::Car(body)), None => Err(BackfillError::Generic(miette::miette!( "getRepo response for {did} exceeded max body size of {max_body_bytes} bytes" ))), }; }
// non-success: decode the (tiny) xrpc error body to preserve typed error handling. let retry_after = (status == StatusCode::TOO_MANY_REQUESTS) .then_some(()) .and_then(|()| parse_retry_after(&resp)); let body = match collect_body_bounded(resp, ERROR_BODY_MAX_BYTES) .await .map_err(|e| BackfillError::Transport(e.to_string().into()))? { Some(body) => body, None => { warn!(%pds, %status, limit = ERROR_BODY_MAX_BYTES, "getRepo error body exceeded limit"); Bytes::new() } };
if let Ok(err) = serde_json::from_slice::<GetRepoError>(&body) { match err { GetRepoError::RepoNotFound(_) => return Ok(FullRepoOutcome::NotFound), GetRepoError::RepoDeactivated(_) => { return Ok(FullRepoOutcome::Inactive(RepoStatus::Deactivated)); } GetRepoError::RepoTakendown(_) => { return Ok(FullRepoOutcome::Inactive(RepoStatus::Takendown)); } GetRepoError::RepoSuspended(_) => { return Ok(FullRepoOutcome::Inactive(RepoStatus::Suspended)); } _ => {} } }
if status == StatusCode::TOO_MANY_REQUESTS { throttle.record_ratelimit(retry_after); return Err(BackfillError::Ratelimited); }
Err(BackfillError::Generic(miette::miette!( "getRepo failed with HTTP {status}: {}", String::from_utf8_lossy(&body) )))}
#[cfg(test)]mod tests { use super::*; use axum::Router; use axum::routing::get;
/// spawns a server answering `com.atproto.sync.getRepo` with a fixed status and body. async fn spawn_get_repo(status: StatusCode, body: Vec<u8>) -> url::Url { let app = Router::new().route( "/xrpc/com.atproto.sync.getRepo", get(move || { let body = body.clone(); async move { (status, body) } }), ); let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); let url = url::Url::parse(&format!("http://{}/", listener.local_addr().unwrap())).unwrap(); tokio::spawn(async move { axum::serve(listener, app).await.unwrap() }); url }
async fn fetch( pds: &url::Url, max_body_bytes: usize, ) -> Result<FullRepoOutcome, BackfillError> { let address = std::net::SocketAddr::new("127.0.0.1".parse().unwrap(), pds.port().unwrap()); let public_pds = url::Url::parse("http://pds.example/").unwrap(); let throttler = crate::util::throttle::Throttler::new(1, 10); let client = reqwest::Client::builder() .resolve("pds.example", address) .build() .unwrap(); let http = ThrottledHttpClient::new( crate::net::PublicHttpClient::test(client), throttler.clone(), ); let throttle = throttler.get_handle(&public_pds).await; let did = Did::new_static("did:plc:aaaaaaaaaaaaaaaaaaaaaaaa").unwrap(); fetch_full_repo_car( &http, &public_pds, &did, &throttle, &RateTier::trusted(), max_body_bytes, ) .await }
#[cfg(feature = "indexer_stream")] #[test] fn imported_identity_event_uses_the_reconciled_state() -> miette::Result<()> { let tmp = tempfile::tempdir().into_diagnostic()?; let app_state = AppState::new(&crate::config::Config { database_path: tmp.path().to_path_buf(), ..Default::default() })?; let did = Did::new_static("did:plc:aaaaaaaaaaaaaaaaaaaaaaaa")?; let mut before = RepoState::synced().into_static(); before.pds = Some("https://old.com".into()); before.handle = Some("old.com".parse()?); let mut after = before.clone(); after.pds = Some("https://new.com".into()); after.handle = Some("new.com".parse()?); let mut events = app_state.db.stream.event_tx.subscribe();
BackfillImported { before, after }.emit(&app_state, &did);
let BroadcastEvent::Ephemeral(event) = events.try_recv().into_diagnostic()? else { panic!("expected ephemeral identity event"); }; let identity = event.identity.expect("identity payload"); assert_eq!(identity.did, did); assert_eq!(identity.handle.as_deref(), Some("new.com")); assert!(events.try_recv().is_err()); Ok(()) }
#[tokio::test] async fn full_get_repo_streams_body_within_ceiling() { let pds = spawn_get_repo(StatusCode::OK, b"car-bytes".to_vec()).await; match fetch(&pds, 1024).await.unwrap() { FullRepoOutcome::Car(body) => assert_eq!(body.as_ref(), b"car-bytes"), other => panic!("expected Car, got {other:?}"), } }
#[tokio::test] async fn full_get_repo_rejects_oversized_body() { let pds = spawn_get_repo(StatusCode::OK, vec![0u8; 4096]).await; let err = fetch(&pds, 64).await.unwrap_err(); assert!( err.to_string().contains("exceeded max body size"), "unexpected error: {err}" ); }
#[tokio::test] async fn full_get_repo_maps_repo_not_found() { let pds = spawn_get_repo( StatusCode::BAD_REQUEST, br#"{"error":"RepoNotFound","message":"nope"}"#.to_vec(), ) .await; assert!(matches!( fetch(&pds, 1024).await.unwrap(), FullRepoOutcome::NotFound )); }
#[tokio::test] async fn full_get_repo_maps_inactive_status() { let pds = spawn_get_repo( StatusCode::BAD_REQUEST, br#"{"error":"RepoDeactivated"}"#.to_vec(), ) .await; match fetch(&pds, 1024).await.unwrap() { FullRepoOutcome::Inactive(status) => assert_eq!(status, RepoStatus::Deactivated), other => panic!("expected Inactive, got {other:?}"), } }}