[backfill] stream getRepo with a hard body bound
102 unmodified linesExpand all103104105106107108109110111112113114115116117203 unmodified linesExpand all321322323324325326327328329330331114 unmodified linesExpand all446447448449450451452453454455456457458103 unmodified linesExpand all102 unmodified linesExpand all pub cursor_save_interval: Duration, /// timeout for fetching a full repository CAR during backfill. /// set via `HYDRANT_REPO_FETCH_TIMEOUT` (humantime duration, e.g. `5min`). pub repo_fetch_timeout: Duration, /// maximum size in bytes of a full-repository `getRepo` CAR response accepted during /// backfill. the body is streamed and rejected before it is fully buffered once it would /// exceed this ceiling, bounding per-task memory regardless of what a PDS returns (the same /// ceiling also bounds sparse `getBlocks` fetches). legitimate repositories are far smaller; /// lower this on memory-constrained hosts running high backfill concurrency. /// set via `HYDRANT_MAX_CAR_BODY_BYTES`. pub max_car_body_bytes: usize, /// maximum number of concurrent backfill tasks. /// set via `HYDRANT_BACKFILL_CONCURRENCY_LIMIT`. pub backfill_concurrency_limit: usize, /// backfill strategy. `full` preserves existing full-repo backfill behavior.203 unmodified linesExpand all firehose_workers: 8, firehose_max_failures: 15, cursor_save_interval: Duration::from_secs(3), repo_fetch_timeout: Duration::from_secs(300), // 2 GiB safety ceiling: never rejects a legitimate repository, only a runaway or // malicious response. memory-constrained/high-concurrency hosts should lower it. max_car_body_bytes: 2 * 1024 * 1024 * 1024, backfill_concurrency_limit: 16, backfill_strategy: BackfillStrategy::Auto, backfill_proxies: vec![], per_pds_concurrency: crate::util::throttle::DEFAULT_PER_PDS_CONCURRENCY_PER_IP,114 unmodified linesExpand all f, "repo fetch timeout", format_args!("{}sec", self.repo_fetch_timeout.as_secs()) )?; config_line!( f, "max car body", format_args!("{} mb", self.max_car_body_bytes / (1024 * 1024)) )?; config_line!(f, "ephemeral", self.ephemeral)?; config_line!(f, "database path", self.database_path.to_string_lossy())?; config_line!(f, "cache size", format_args!("{} mb", self.cache_size))?; config_line!(f, "data compression", self.data_compression)?;103 unmodified linesExpand all42 unmodified lines43444546474849505172 unmodified lines12412512612712812913013113298 unmodified lines42 unmodified lines pub ephemeral: bool, pub ephemeral_ttl: Duration, pub only_index_links: bool, pub verify_cids: bool, pub max_car_body_bytes: usize, pub get_repo_semaphore: Semaphore,}
impl AppState {72 unmodified lines ephemeral: config.ephemeral, ephemeral_ttl: config.ephemeral_ttl, only_index_links: config.only_index_links, verify_cids: config.verify_cids, max_car_body_bytes: config.max_car_body_bytes, get_repo_semaphore: Semaphore::new(config.get_repo_concurrency_limit), throttler: Throttler::new( config.backfill_proxies.len() + 1, config.per_pds_concurrency,98 unmodified lines12345683 unmodified lines90919293949596979899100101102103104105106107108109110111112113114115116117118119120121use crate::util::throttle::Throttler;use bytes::{Bytes, BytesMut};use jacquard_common::http_client::HttpClient;use reqwest::StatusCode;use std::sync::Arc;use std::sync::atomic::{AtomicUsize, Ordering};83 unmodified lines
res }}
/// streams a response body into a single contiguous [`Bytes`], enforcing a hard byte ceiling.////// returns `Ok(None)` when the body exceeds `max_bytes`: the response is dropped without/// buffering the remainder, so per-task memory stays bounded regardless of what the PDS sends/// (a chunked response with no `Content-Length`, or a decompression bomb — `chunk` yields/// already-decoded bytes, so the ceiling applies to the decompressed size). the buffer is grown/// in place and never double-buffered: at most `buf + one chunk` is resident at any point.pub(crate) async fn collect_body_bounded( mut resp: reqwest::Response, max_bytes: usize,) -> Result<Option<Bytes>, reqwest::Error> { // reject early when a declared (uncompressed) length already exceeds the ceiling. reqwest // strips `Content-Length` from decoded responses, so a present value is the real body size. let content_length = resp.content_length(); if content_length.is_some_and(|len| len > max_bytes as u64) { return Ok(None); } let reserve = content_length.unwrap_or(0).min(max_bytes as u64) as usize; let mut buf = BytesMut::with_capacity(reserve); while let Some(chunk) = resp.chunk().await? { if buf.len() + chunk.len() > max_bytes { return Ok(None); } buf.extend_from_slice(&chunk); } Ok(Some(buf.freeze()))}112345192 unmodified linesExpand all19819920020120220320420520636 unmodified lines24324424524624724824925025136 unmodified lines28828929029129229329429529629729529829930030130230330430530630730830931031131231311 unmodified lines32532632732832933033133233335 unmodified lines369370371372358359373374361375376377378379380381382383384293 unmodified linesExpand all67867968068168268368468568640 unmodified lines7277287297307317327337347355 unmodified lines741742743744745746747748749750751752753732754755756757758759760761762763764741765766767768769770771772773774749775776777778779780781782783784785786787788use crate::backfill::client::ThrottledHttpClient;use crate::backfill::client::{ThrottledHttpClient, collect_body_bounded};use crate::backfill::error::BackfillError;use crate::config::{BackfillStrategy, RateTier};use crate::db::types::{DbAction, DbRkey};use crate::db::{self, Txn as DbTxn, keys};192 unmodified linesExpand all &missing, &throttle, &tier, app_state.verify_cids, app_state.max_car_body_bytes, ) .await?; if missing.iter().all(|cid| !blocks.contains_key(cid)) { return Ok(SparseBackfillResult::Skipped);36 unmodified lines &missing_records, &throttle, &tier, app_state.verify_cids, app_state.max_car_body_bytes, ) .await?, );
36 unmodified lines cids: &[IpldCid], throttle: &ThrottleHandle, tier: &RateTier, verify_cids: bool, max_body_bytes: usize,) -> Result<BTreeMap<IpldCid, bytes::Bytes>, BackfillError> { let mut out = BTreeMap::new();
let fetches = cids .chunks(SPARSE_GET_BLOCKS_CHUNK) .map(|chunk| fetch_block_chunk(http, pds, did, chunk, throttle, tier, verify_cids)) .map(|chunk| { fetch_block_chunk( http, pds, did, chunk, throttle, tier, verify_cids, max_body_bytes, ) }) .collect::<Vec<_>>(); let fetches = stream::iter(fetches).buffer_unordered(SPARSE_GET_BLOCKS_PARALLELISM); futures::pin_mut!(fetches);
11 unmodified lines cids: &[IpldCid], throttle: &ThrottleHandle, tier: &RateTier, verify_cids: bool, max_body_bytes: usize,) -> Result<BTreeMap<IpldCid, bytes::Bytes>, BackfillError> { let mut url = pds .join("xrpc/com.atproto.sync.getBlocks") .map_err(|e| BackfillError::Generic(miette::miette!("Invalid URL: {e}")))?;35 unmodified lines };
let status = resp.status(); if status.is_success() { let body = resp .bytes() let Some(body) = collect_body_bounded(resp, max_body_bytes) .await .map_err(|e| BackfillError::Transport(e.to_string().into()))?; .map_err(|e| BackfillError::Transport(e.to_string().into()))? else { return Err(BackfillError::Generic(miette::miette!( "getBlocks response for {did} exceeded max body size of {max_body_bytes} bytes" ))); }; let blocks = crate::car::parse_car_blocks(body).map_err(BackfillError::from)?; if verify_cids { crate::car::validate_block_cids(&blocks)?; }293 unmodified linesExpand all &cids, &throttle, &RateTier::trusted(), false, 64 * 1024 * 1024, ), ) .await .expect("getBlocks chunks were fetched serially")40 unmodified lines async fn fetch_one( pds: &url::Url, wanted: IpldCid, verify_cids: bool, max_body_bytes: usize, ) -> Result<BTreeMap<IpldCid, bytes::Bytes>, BackfillError> { let throttler = crate::util::throttle::Throttler::new(1, 10); let http = ThrottledHttpClient::new(vec![reqwest::Client::new()], throttler.clone()); let throttle = throttler.get_handle(pds).await;5 unmodified lines &[wanted], &throttle, &RateTier::trusted(), verify_cids, max_body_bytes, ) .await }
#[tokio::test] async fn verify_cids_rejects_mismatched_get_blocks_car() { let pds = spawn_car_server(car_with_block(cid(1), b"forged").await).await;
let err = fetch_one(&pds, cid(1), true).await.unwrap_err(); let err = fetch_one(&pds, cid(1), true, 64 * 1024 * 1024) .await .unwrap_err(); assert!(err.to_string().contains("CAR block CID mismatch")); }
#[tokio::test] async fn verify_cids_accepts_matching_get_blocks_car() { let real_cid = jacquard_repo::mst::util::compute_cid(b"trusted").unwrap(); let pds = spawn_car_server(car_with_block(real_cid, b"trusted").await).await;
let blocks = fetch_one(&pds, real_cid, true).await.unwrap(); let blocks = fetch_one(&pds, real_cid, true, 64 * 1024 * 1024) .await .unwrap(); assert_eq!(blocks.get(&real_cid).unwrap().as_ref(), b"trusted"); }
#[tokio::test] async fn verify_cids_disabled_accepts_mismatched_get_blocks_car() { let pds = spawn_car_server(car_with_block(cid(1), b"forged").await).await;
let blocks = fetch_one(&pds, cid(1), false).await.unwrap(); let blocks = fetch_one(&pds, cid(1), false, 64 * 1024 * 1024) .await .unwrap(); assert_eq!(blocks.get(&cid(1)).unwrap().as_ref(), b"forged"); }
#[tokio::test] async fn get_blocks_rejects_oversized_body() { let pds = spawn_car_server(car_with_block(cid(1), &[0u8; 4096]).await).await;
let err = fetch_one(&pds, cid(1), false, 64).await.unwrap_err(); assert!(err.to_string().contains("exceeded max body size")); }}117 unmodified linesExpand all118119120121122123124125126233 unmodified linesExpand all36036136236336436536636736836 unmodified lines117 unmodified linesExpand all .unwrap_or_else(|| Ok(defaults.plc_urls.clone()))?;
let cursor_save_interval = cfg!("CURSOR_SAVE_INTERVAL", defaults.cursor_save_interval, sec); let repo_fetch_timeout = cfg!("REPO_FETCH_TIMEOUT", defaults.repo_fetch_timeout, sec); let max_car_body_bytes = cfg!("MAX_CAR_BODY_BYTES", defaults.max_car_body_bytes);
let ephemeral: bool = cfg!("EPHEMERAL", defaults.ephemeral); let ephemeral_ttl = cfg!("EPHEMERAL_TTL", defaults.ephemeral_ttl, sec); let database_path = cfg!("DATABASE_PATH", defaults.database_path);233 unmodified linesExpand all firehose_max_failures, cursor_save_interval, enable_backfill, repo_fetch_timeout, max_car_body_bytes, backfill_concurrency_limit, backfill_strategy, backfill_proxies, per_pds_concurrency,36 unmodified lines123456789101110121314151416171818192021212223242526272829293031323334159 unmodified linesExpand all1941951961971971981992002011992032042052062072082092102112122132142152002012022032042052062072082092102112122132142152162172182192202172212192222232242252212262272282232242252262272282292302312322332342352362372382392402292302312422432442452462322332342352362492502512522532542552562572582592602612622632642652662672372382392402412422432442452462472482492502512522532542692702552562572582592602612722732742752762772782792802812822832842852862872882892622632642652662672912922932942682692962702712722734 unmodified lines278279280281309282283284285286314315316317318287288321289290291292293228 unmodified linesExpand all522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728use 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, error, trace, warn};
use jacquard_api::com_atproto::sync::get_repo::{GetRepo, GetRepoError};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_common::xrpc::{XrpcError, XrpcExt};use jacquard_repo::mst::Mst;use jacquard_repo::{BlockStore, MemoryBlockStore};
use crate::backfill::client::ThrottledHttpClient;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;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::ops;use crate::sparse_mst::sparse_probe_collection;use crate::state::AppState;use crate::types::{Commit, GaugeState, RepoState, RepoStatus, ResyncState};use crate::util::url_to_fluent_uri;use crate::util::{parse_retry_after, throttle::ThrottleHandle};
#[cfg(feature = "indexer_stream")]use crate::types::{AccountEvt, BroadcastEvent};#[cfg(feature = "indexer_stream")]159 unmodified linesExpand all }
// 2. fetch repo (car) let start = Instant::now(); let req = GetRepo::new().did(did.clone()).build(); let throttle = app_state.throttler.get_handle(&pds).await; if throttle.is_throttled() { return Err(BackfillError::PreemptivelyThrottled); } let tier = app_state.resolve_pds_tier(pds.host_str().unwrap_or("")); let resp = { 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); } match http.xrpc(url_to_fluent_uri(&pds)).send(&req).await { Ok(resp) => { if !throttle.is_throttled() { throttle.record_success(); 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 => { warn!("repo not found, deleting"); let mut txn = DbTxn::new(db); let applied = txn.transition_pending_key(did, pending_key.as_ref(), GaugeState::Synced)?; txn.batch.remove(&db.indexer.pending, pending_key.clone()); if applied { if let Err(e) = crate::ops::delete_repo(&mut txn.batch, db, did, &state) { error!(err = %e, "failed to wipe repo during backfill"); } resp } Err(e) => return Err(BackfillError::from_sparse_client(e, &throttle)), txn.commit()?; // return None so did_task skips sending BackfillFinished (nothing to drain for a deleted repo) return Ok(None); } }; FullRepoOutcome::Inactive(status) => { warn!(?status, "repo is inactive, stopping backfill");
let car_bytes = match resp.into_output() { Ok(o) => o, Err(XrpcError::Xrpc(e)) => { if matches!(e, GetRepoError::RepoNotFound(_)) { warn!("repo not found, deleting"); let mut txn = DbTxn::new(db); let applied = txn.transition_pending_key(did, pending_key.as_ref(), GaugeState::Synced)?; txn.batch.remove(&db.indexer.pending, pending_key.clone()); if applied { if let Err(e) = crate::ops::delete_repo(&mut txn.batch, db, did, &state) { error!(err = %e, "failed to wipe repo during backfill"); } } txn.commit()?; // return None so did_task skips sending BackfillFinished (nothing to drain for a deleted repo) return Ok(None); } #[cfg(feature = "indexer_stream")] emit_identity(&status, false);
let inactive_status = match e { GetRepoError::RepoDeactivated(_) => Some(RepoStatus::Deactivated), GetRepoError::RepoTakendown(_) => Some(RepoStatus::Takendown), GetRepoError::RepoSuspended(_) => Some(RepoStatus::Suspended), _ => None, let resync_state = ResyncState::Gone { status: status.clone(), }; let resync_bytes = rmp_serde::to_vec(&resync_state).into_diagnostic()?;
if let Some(status) = inactive_status { warn!(?status, "repo is inactive, stopping backfill");
#[cfg(feature = "indexer_stream")] emit_identity(&status, false);
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(); let did = did.clone(); let pending_key = pending_key.clone(); app_state_clone .db .run(move |db| { let mut txn = DbTxn::new(db); let applied = txn.transition_pending_key( let app_state_clone = app_state.clone(); let did = did.clone(); let pending_key = pending_key.clone(); app_state_clone .db .run(move |db| { let mut txn = DbTxn::new(db); 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, pending_key.as_ref(), GaugeState::Resync(None), move |state, (key, batch)| { state.active = false; state.status = status; batch.insert(&db.indexer.resync, key, resync_bytes); Ok((true, ())) }, )?; 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>(()) }) .await?; } txn.commit()?; Ok::<_, miette::Report>(()) }) .await?;
return Ok(None); }
Err(e).into_diagnostic()? return Ok(None); } Err(e) => Err(e).into_diagnostic()?, };
// emit identity event so any consumers know, but only if something changed #[cfg(feature = "indexer_stream")]4 unmodified lines emit_identity(&state.status, state.active); }
trace!( bytes = car_bytes.body.len(), bytes = car_bytes.len(), elapsed = ?start.elapsed(), "fetched car bytes" );
// TODO: enforce a max_car_body_bytes limit here before parsing to prevent a malicious // or compromised PDS from returning an unbounded response and causing OOM. // also apply to the sparse path in sparse.rs. // (finding d87fbd22402c81919fceb9cdfee53cd4)
// 3. import repo let start = Instant::now(); let parsed = crate::car::parse_car(car_bytes.body.into())?; 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;228 unmodified linesExpand all
trace!("complete"); Ok(Some(previous_state))}
/// 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 reported that the repository does not exist (`RepoNotFound`). 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 mut url = pds .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());
// 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 resp = match http .get(url) .header(reqwest::header::ACCEPT, "application/vnd.ipld.car") .send() .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(|| parse_retry_after(&resp)) .flatten(); let body = collect_body_bounded(resp, ERROR_BODY_MAX_BYTES) .await .map_err(|e| BackfillError::Transport(e.to_string().into()))? .unwrap_or_default();
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 throttler = crate::util::throttle::Throttler::new(1, 10); let http = ThrottledHttpClient::new(vec![reqwest::Client::new()], throttler.clone()); let throttle = throttler.get_handle(pds).await; let did = Did::new_static("did:plc:aaaaaaaaaaaaaaaaaaaaaaaa").unwrap(); fetch_full_repo_car( &http, pds, &did, &throttle, &RateTier::trusted(), max_body_bytes, ) .await }
#[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:?}"), } }}