Something went wrong. Try again.
fork of hydrant
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819820821822823824825826827828use crate::backfill::client::ThrottledHttpClient;use crate::backfill::error::BackfillError;use crate::config::{BackfillStrategy, RateTier};use crate::db::types::{DbAction, DbRkey};use crate::db::{self, Txn as DbTxn, keys};use crate::filter::{FilterConfig, FilterMode};use crate::net::PublicEndpoint;use crate::ops;use crate::sparse_mst::{SparseScanner, mst_node_layer, sparse_probe_collection, sparse_ranges};use crate::state::AppState;use crate::types::{Commit, RepoState};
use futures::{StreamExt, stream};use jacquard_api::com_atproto::sync::get_blocks::GetBlocksError;use jacquard_api::com_atproto::sync::get_record::{GetRecord, GetRecordError};use jacquard_common::http_client::HttpClient;use jacquard_common::types::cid::{Cid as AtCid, IpldCid};use jacquard_common::types::did::Did;use jacquard_common::types::string::{Nsid, RecordKey};use jacquard_common::xrpc::{XrpcError, XrpcExt};use miette::IntoDiagnostic;use reqwest::StatusCode;
use smol_str::{SmolStr, ToSmolStr};use std::collections::{BTreeMap, HashMap};use std::sync::Arc;use tracing::{debug, trace, warn};
use crate::util::throttle::ThrottleHandle;use crate::util::url_to_fluent_uri_ref;
#[derive(Debug)]pub(crate) struct SparseBackfillSuccess { pub(crate) state: RepoState<'static>, pub(crate) records: usize, pub(crate) node_blocks: usize, pub(crate) node_bytes: usize,}
#[derive(Debug)]pub(crate) enum SparseBackfillResult { Imported(SparseBackfillSuccess), Discarded, Skipped,}
const SPARSE_GET_BLOCKS_CHUNK: usize = 100;const SPARSE_GET_BLOCKS_PARALLELISM: usize = 4;const SPARSE_MAX_SCAN_ROUNDS: usize = 256;const SPARSE_AUTO_FULL_MAX_ROOT_LAYER: usize = 2;fn sparse_probe(filter: &FilterConfig) -> Option<(SmolStr, &'static str)> { filter .collections .iter() .find(|collection| !collection.ends_with(".*") && collection.ends_with(".profile")) .or_else(|| { filter .signals .iter() .find(|signal| signal.ends_with(".profile") && filter.matches_collection(signal)) }) .cloned() .map(|collection| (collection, "self")) .or_else(|| { sparse_probe_collection(&filter.collections).map(|collection| (collection, "-")) })}
pub(crate) async fn process_did_sparse( app_state: &Arc<AppState>, http: &ThrottledHttpClient, did: &Did, pds: &url::Url, state: RepoState<'static>, verify_signatures: bool, strategy: BackfillStrategy,) -> Result<SparseBackfillResult, BackfillError> { let pds_endpoint = PublicEndpoint::parse_http(pds).map_err(|error| { BackfillError::Generic(miette::miette!("unsafe public PDS endpoint {pds}: {error}")) })?; let filter = app_state.filter.load(); let Some((probe_collection, probe_rkey)) = sparse_probe(&filter) else { return Ok(SparseBackfillResult::Skipped); }; let ranges = sparse_ranges(&filter.collections); if ranges.is_empty() { return Ok(SparseBackfillResult::Skipped); }
let probe_collection = Nsid::new_owned(probe_collection.as_str()).into_diagnostic()?; let probe_rkey = RecordKey::any_static(probe_rkey).into_diagnostic()?; let req = GetRecord::new() .did(did.clone()) .collection(probe_collection) .rkey(probe_rkey) .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_ref(pds_endpoint.url())) .send(&req) .await { Ok(resp) => { if !throttle.is_throttled() { throttle.record_success(); } resp } Err(e) => return Err(BackfillError::from_sparse_client(e, &throttle)), } }; let proof = match resp.into_output() { Ok(o) => o, Err(XrpcError::Xrpc(GetRecordError::RecordNotFound(_))) => { return Ok(SparseBackfillResult::Skipped); } Err(XrpcError::Xrpc( GetRecordError::RepoNotFound(_) | GetRecordError::RepoTakendown(_) | GetRecordError::RepoSuspended(_) | GetRecordError::RepoDeactivated(_), )) => return Ok(SparseBackfillResult::Skipped), Err(e) => Err(e).into_diagnostic()?, };
let parsed = jacquard_repo::car::reader::parse_car_bytes(&proof.body) .await .into_diagnostic()?; if app_state.verify_cids { crate::car::validate_block_cids(&parsed.blocks)?; } let root_bytes = parsed .blocks .get(&parsed.root) .ok_or_else(|| miette::miette!("root block missing from sparse proof CAR"))?; let root_commit = jacquard_repo::commit::Commit::<jacquard_common::DefaultStr>::from_cbor(root_bytes) .into_diagnostic()?;
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}"))?; }
let root_cid = root_commit.data; let auto_full_layer = if strategy == BackfillStrategy::Auto { parsed .blocks .get(&root_cid) .map(|root_bytes| mst_node_layer(root_bytes)) .transpose()? .flatten() .filter(|layer| *layer <= SPARSE_AUTO_FULL_MAX_ROOT_LAYER) } else { None };
let root_commit = Commit::from(root_commit); let mut scanner = SparseScanner::new(ranges, parsed.blocks); let mut scan_rounds = 0; let scan = loop { match scanner.scan(root_cid)? { Ok(scan) => break scan, Err(missing) => { if let Some(root_layer) = auto_full_layer { debug!( root_layer, max_sparse_layer = SPARSE_AUTO_FULL_MAX_ROOT_LAYER, "sparse auto selected full getRepo because the probe omitted MST nodes" ); return Ok(SparseBackfillResult::Skipped); } scan_rounds += 1; if scan_rounds > SPARSE_MAX_SCAN_ROUNDS { return Err( miette::miette!("sparse mst scan exceeded fetch round limit").into(), ); } if missing.is_empty() { return Ok(SparseBackfillResult::Skipped); } if missing.len() > SPARSE_MAX_SCAN_ROUNDS * SPARSE_GET_BLOCKS_CHUNK { return Err( miette::miette!("sparse mst scan exceeded missing block limit").into(), ); } let blocks = fetch_blocks( http, pds, did, &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); } scanner.insert_blocks(blocks); } } };
let record_cids = scan .leaves .iter() .map(|(_, cid)| *cid) .collect::<std::collections::BTreeSet<_>>(); let mut scanned_blocks = scanner.take_blocks(); let missing_records = record_cids .iter() .filter_map(|cid| (!scanned_blocks.contains_key(cid)).then_some(*cid)) .collect::<Vec<_>>(); if !missing_records.is_empty() && let Some(root_layer) = auto_full_layer { debug!( root_layer, max_sparse_layer = SPARSE_AUTO_FULL_MAX_ROOT_LAYER, "sparse auto selected full getRepo because the probe omitted record blocks" ); return Ok(SparseBackfillResult::Skipped); } let mut record_blocks = record_cids .iter() .filter_map(|cid| scanned_blocks.remove(cid).map(|bytes| (*cid, bytes))) .collect::<BTreeMap<_, _>>(); drop(scanned_blocks); record_blocks.extend( fetch_blocks( http, pds, did, &missing_records, &throttle, &tier, app_state.verify_cids, app_state.max_car_body_bytes, ) .await?, );
if let Some((_, missing)) = scan .leaves .iter() .find(|(_, cid)| !record_blocks.contains_key(cid)) { return Err( miette::miette!("sparse record block missing after getBlocks: {missing}").into(), ); }
let result = persist_sparse_backfill( app_state, did, state, root_commit, scan.leaves, record_blocks, ) .await?;
let Some((records, state)) = result else { return Ok(SparseBackfillResult::Discarded); };
Ok(SparseBackfillResult::Imported(SparseBackfillSuccess { state, records, node_blocks: scan.node_blocks_seen, node_bytes: scan.node_bytes_seen, }))}
async fn fetch_blocks( http: &ThrottledHttpClient, pds: &url::Url, did: &Did, 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 = stream::iter((0..cids.len()).step_by(SPARSE_GET_BLOCKS_CHUNK)) .map(|start| { let end = start .saturating_add(SPARSE_GET_BLOCKS_CHUNK) .min(cids.len()); fetch_block_chunk( http, pds, did, &cids[start..end], throttle, tier, verify_cids, max_body_bytes, ) }) .buffer_unordered(SPARSE_GET_BLOCKS_PARALLELISM); futures::pin_mut!(fetches);
while let Some(blocks) = fetches.next().await { out.extend(blocks?); }
Ok(out)}
async fn fetch_block_chunk( http: &ThrottledHttpClient, pds: &url::Url, did: &Did, 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}")))?; { let mut query = url.query_pairs_mut(); query.append_pair("did", did.as_str()); for cid in cids { query.append_pair("cids", &cid.to_string()); } }
let endpoint = PublicEndpoint::parse_http(&url).map_err(|error| { BackfillError::Generic(miette::miette!( "unsafe public getBlocks endpoint {url}: {error}" )) })?; let resp = { let _permit = throttle.acquire().await; throttle.wait_for_allow(1, tier).await; let request = http::Request::builder() .method(http::Method::GET) .uri(endpoint.url().as_str()) .header(http::header::ACCEPT, "application/vnd.ipld.car") .body(Vec::new()) .map_err(|e| { BackfillError::Generic(miette::miette!("invalid getBlocks request: {e}")) })?; match HttpClient::send_http(http, 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() { let body = resp.into_body(); if body.len() > max_body_bytes { 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(bytes::Bytes::from(body)).map_err(BackfillError::from)?; if verify_cids { crate::car::validate_block_cids(&blocks)?; } Ok(blocks) } else { let retry_after = (status == StatusCode::TOO_MANY_REQUESTS) .then_some(()) .and_then(|()| { resp.headers() .get(http::header::RETRY_AFTER) .and_then(|value| value.to_str().ok()) .and_then(|value| value.parse::<u64>().ok()) });
let body = resp.into_body(); if serde_json::from_slice::<GetBlocksError>(&body) .ok() .is_some_and(|err| { matches!( err, GetBlocksError::BlockNotFound(_) | GetBlocksError::RepoNotFound(_) | GetBlocksError::RepoTakendown(_) | GetBlocksError::RepoSuspended(_) | GetBlocksError::RepoDeactivated(_) ) }) { return Ok(BTreeMap::new()); }
if status == StatusCode::TOO_MANY_REQUESTS { throttle.record_ratelimit(retry_after); return Err(BackfillError::Ratelimited); }
Err(miette::miette!( "getBlocks failed with HTTP {status}: {}", String::from_utf8_lossy(&body) ) .into()) }}
async fn persist_sparse_backfill( app_state: &Arc<AppState>, did: &Did, mut state: RepoState<'static>, root_commit: Commit, leaves: Vec<(SmolStr, IpldCid)>, blocks: BTreeMap<IpldCid, bytes::Bytes>,) -> Result<Option<(usize, RepoState<'static>)>, BackfillError> { let app_state = app_state.clone(); let did = did.clone(); let db = app_state.db.clone(); db.run(move |db| { let filter = app_state.filter.load(); let mut count = 0; let mut collection_counts: HashMap<SmolStr, u64> = HashMap::new();
let mut txn = DbTxn::new(db); // permanent mode locks before the existing-head snapshot so a live // write or operator redaction cannot invalidate the diff. ephemeral // mode has no heads and therefore no shared record state to lock. if !app_state.ephemeral { txn.hold_repo_write_lock(&did); }
let prefix = keys::record_prefix_did(&did); let mut existing_cids: HashMap<(SmolStr, DbRkey), SmolStr> = HashMap::new();
if !app_state.ephemeral { for guard in 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}") })?; if !filter.matches_collection(collection) { continue; }
// 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(); let mut record_txn = txn.backfill_records(&app_state, &root_commit.rev, &did)?; if record_txn.is_excluded() { return Ok::<_, miette::Report>(None); }
for (key, cid) in leaves { let (collection, rkey) = ops::parse_path(&key)?;
if !filter.matches_collection(collection) { continue; }
let Some(val) = blocks.get(&cid) else { return Err(miette::miette!("missing sparse record block {cid}")); };
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 sparse record"); continue; }; trace!(collection = %collection, rkey = %rkey, cid = %cid, ?action, "action sparse record");
record_txn.put_record(collection, &rkey, &cid, val, action)?; count += 1; }
for ((collection, rkey), cid) in existing_cids { trace!(collection = %collection, rkey = %rkey, cid = %cid, "remove sparse-stale record"); record_txn.delete_record(&collection, &rkey)?; count += 1; }
if !signal_seen { trace!(signals = ?filter.signals, "no signal-matching sparse records found, discarding repo"); return Ok::<_, miette::Report>(None); }
state.root = Some(root_commit); state.touch(); record_txn.update_repo_state(&state)?; let _events = record_txn.finish()?;
let metadata_key = keys::repo_metadata_key(&did); let metadata_bytes = 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( &db.repo_metadata, &metadata_key, crate::db::ser_repo_meta(&metadata)?, );
db::replace_record_counts_matching( &mut txn.batch, db, &did, |collection| filter.matches_collection(collection), collection_counts.iter().map(|(col, cnt)| (col.as_str(), *cnt)), )?;
txn.commit()?;
Ok::<_, miette::Report>(Some((count, state))) }) .await .map_err(BackfillError::from)}
#[cfg(test)]mod tests { use super::*; use crate::filter::FilterConfig; use axum::{Router, extract::State, response::IntoResponse, routing::get}; use cid::Cid; use cid::multihash::Multihash; use jacquard_common::types::crypto::{DAG_CBOR, SHA2_256}; use std::sync::atomic::{AtomicUsize, Ordering}; use std::time::Duration; use tokio::sync::Barrier;
#[derive(Clone)] struct FetchState { started: Arc<AtomicUsize>, active: Arc<AtomicUsize>, max_active: Arc<AtomicUsize>, barrier: Arc<Barrier>, car: bytes::Bytes, }
async fn delayed_car(State(state): State<FetchState>) -> impl IntoResponse { let ordinal = state.started.fetch_add(1, Ordering::SeqCst) + 1; let active = state.active.fetch_add(1, Ordering::SeqCst) + 1; state.max_active.fetch_max(active, Ordering::SeqCst); if ordinal <= 2 { state.barrier.wait().await; } state.active.fetch_sub(1, Ordering::SeqCst); ( [( reqwest::header::CONTENT_TYPE.as_str(), "application/vnd.ipld.car", )], state.car, ) }
fn cid(byte: u8) -> IpldCid { let hash = [byte; 32]; let multihash = Multihash::<64>::wrap(SHA2_256, &hash).unwrap(); Cid::new_v1(DAG_CBOR, multihash) }
fn filter(collections: &[&str], signals: &[&str]) -> FilterConfig { let mut filter = FilterConfig::new(FilterMode::Filter); filter.collections = collections.iter().map(SmolStr::new).collect(); filter.signals = signals.iter().map(SmolStr::new).collect(); filter }
#[test] fn sparse_probe_prefers_matching_profile_signal() { let filter = filter( &["sh.tangled.*"], &["sh.tangled.actor.profile", "app.bsky.actor.profile"], );
assert_eq!( sparse_probe(&filter), Some((SmolStr::new("sh.tangled.actor.profile"), "self")) ); }
#[test] fn sparse_probe_ignores_profile_signal_outside_filter() { let filter = filter(&["sh.tangled.*"], &["app.bsky.actor.profile"]);
assert_eq!( sparse_probe(&filter), Some((SmolStr::new("sh.tangled.probe"), "-")) ); }
#[test] fn sparse_probe_uses_self_for_exact_profile_collection() { let filter = filter(&["app.bsky.actor.profile"], &[]);
assert_eq!( sparse_probe(&filter), Some((SmolStr::new("app.bsky.actor.profile"), "self")) ); }
#[tokio::test] async fn fetches_independent_get_blocks_chunks_concurrently() { let mut car = Vec::new(); let mut writer = iroh_car::CarWriter::new(iroh_car::CarHeader::new_v1(Vec::new()), &mut car); let response_cid = cid(u8::MAX); writer.write(response_cid, b"block".to_vec()).await.unwrap(); writer.finish().await.unwrap(); let state = FetchState { started: Arc::new(AtomicUsize::new(0)), active: Arc::new(AtomicUsize::new(0)), max_active: Arc::new(AtomicUsize::new(0)), barrier: Arc::new(Barrier::new(2)), car: car.into(), }; let app = Router::new() .route("/xrpc/com.atproto.sync.getBlocks", get(delayed_car)) .with_state(state.clone()); let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); let address = listener.local_addr().unwrap(); let pds = url::Url::parse("http://pds.example/").unwrap(); tokio::spawn(async move { axum::serve(listener, app).await.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(&pds).await; let did = Did::new_static("did:plc:aaaaaaaaaaaaaaaaaaaaaaaa").unwrap(); let cids = (0..=200).map(|byte| cid(byte as u8)).collect::<Vec<_>>();
let blocks = tokio::time::timeout( Duration::from_secs(5), fetch_blocks( &http, &pds, &did, &cids, &throttle, &RateTier::trusted(), false, 64 * 1024 * 1024, ), ) .await .expect("getBlocks chunks were fetched serially") .unwrap();
assert_eq!(blocks.get(&response_cid).unwrap().as_ref(), b"block"); assert!( state.max_active.load(Ordering::SeqCst) > 1, "expected more than one getBlocks chunk in flight" ); }
async fn spawn_car_server(car: bytes::Bytes) -> url::Url { let app = Router::new().route( "/xrpc/com.atproto.sync.getBlocks", get(move || { let car = car.clone(); async move { ( [( reqwest::header::CONTENT_TYPE.as_str(), "application/vnd.ipld.car", )], car, ) } }), ); 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 car_with_block(cid: IpldCid, payload: &[u8]) -> bytes::Bytes { let mut car = Vec::new(); let mut writer = iroh_car::CarWriter::new(iroh_car::CarHeader::new_v1(Vec::new()), &mut car); writer.write(cid, payload.to_vec()).await.unwrap(); writer.finish().await.unwrap(); car.into() }
async fn fetch_one( pds: &url::Url, wanted: IpldCid, verify_cids: bool, max_body_bytes: usize, ) -> Result<BTreeMap<IpldCid, bytes::Bytes>, 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_blocks( &http, &public_pds, &did, &[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, 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, 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, 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")); }}