use crate::backfill::client::ThrottledHttpClient;
use crate::backfill::error::BackfillError;
use crate::backfill::worker::records::FilteredRecords;
use crate::car::{Blocks, CommitBlocks, CommitCar};
use crate::config::{BackfillStrategy, RateTier};
use crate::filter::FilterConfig;
use crate::net::PublicEndpoint;
use crate::sparse_mst::{SparseScanner, mst_node_layer, sparse_probe_collection, sparse_ranges};
use crate::state::AppState;
use crate::types::Commit;
use futures::{StreamExt, stream};
use jacquard_common::types::cid::{Cid, IpldCid};
use jacquard_common::types::did::Did;
use jacquard_common::types::string::{Nsid, Rkey};
use miette::IntoDiagnostic;
use reqwest::StatusCode;
use serde::Deserialize;
use smol_str::SmolStr;
use std::sync::Arc;
use tracing::{debug, warn};
use crate::util::throttle::ThrottleHandle;
#[derive(Deserialize)]
struct XrpcErrorEnvelope {
error: SmolStr,
}
#[derive(Deserialize)]
struct LatestCommitOutput {
cid: Cid,
}
pub(crate) fn xrpc_error_code(body: &[u8]) -> Option<SmolStr> {
serde_json::from_slice::<XrpcErrorEnvelope>(body)
.ok()
.map(|err| err.error)
}
/// what a sparse fetch found, for the backfill to write through the repo's shard like a
/// getRepo's records
pub(crate) struct SparseBackfillSuccess {
pub(crate) root_commit: Commit,
pub(crate) records: FilteredRecords,
/// the filter whose collections it fetched, see
/// [`crate::ingest::indexer::message::BackfillTail::fetched_only`]
pub(crate) filter: Arc<FilterConfig>,
pub(crate) node_blocks: usize,
pub(crate) node_bytes: usize,
}
pub(crate) enum SparseBackfillResult {
Fetched(Box<SparseBackfillSuccess>),
Skipped,
}
const SPARSE_GET_BLOCKS_CHUNK: usize = 100;
const SPARSE_GET_BLOCKS_PARALLELISM: usize = 4;
pub(crate) 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, "-"))
})
}
#[allow(clippy::too_many_arguments)]
pub(crate) async fn process_did_sparse(
app_state: &Arc<AppState>,
http: &ThrottledHttpClient,
did: &Did,
pds: &url::Url,
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_full();
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 = Nsid::new_owned(probe_collection.as_str()).into_diagnostic()?;
let probe_rkey: Rkey = Rkey::new_static(probe_rkey).into_diagnostic()?;
let mut record_url = pds_endpoint
.url()
.join("xrpc/com.atproto.sync.getRecord")
.map_err(|e| BackfillError::Generic(miette::miette!("Invalid URL: {e}")))?;
record_url
.query_pairs_mut()
.append_pair("collection", probe_collection.as_str())
.append_pair("did", did.as_str())
.append_pair("rkey", probe_rkey.as_ref());
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 = send_throttled(
http,
&record_url,
"application/vnd.ipld.car",
&throttle,
&tier,
)
.await?;
let status = resp.status();
let body = resp.into_body();
let start = if status.is_success() {
Some(crate::car::parse_commit_car(
bytes::Bytes::from(body),
app_state.verify_cids,
)?)
} else if status == StatusCode::NOT_FOUND {
None
} else if status == StatusCode::BAD_REQUEST {
match xrpc_error_code(&body).as_deref() {
Some("RecordNotFound") => None,
Some("RepoNotFound" | "RepoTakendown" | "RepoSuspended" | "RepoDeactivated") => {
return Ok(SparseBackfillResult::Skipped);
}
_ => {
return Err(miette::miette!(
"getRecord failed with HTTP {status}: {}",
String::from_utf8_lossy(&body)
)
.into());
}
}
} else if status == StatusCode::TOO_MANY_REQUESTS {
return Err(BackfillError::Ratelimited);
} else {
return Err(miette::miette!(
"getRecord failed with HTTP {status}: {}",
String::from_utf8_lossy(&body)
)
.into());
};
let start = match start {
Some(proof) => proof,
None => {
debug!("sparse probe has no proof, starting from the latest commit");
let latest_commit =
latest_commit_start(app_state, http, did, pds, &pds_endpoint, &throttle, &tier)
.await?;
let Some(start) = latest_commit else {
return Ok(SparseBackfillResult::Skipped);
};
start
}
};
let root_bytes = start
.blocks
.get(&start.root)
.ok_or_else(|| miette::miette!("root block missing from sparse proof CAR"))?;
let root_commit = crate::mst::AtpCommit::from_cbor(root_bytes)?;
check_commit_did(root_commit.did.as_str(), did)?;
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 mut blocks = start.blocks;
// a start from the latest commit has no mst nodes yet. fetching the root now is the
// scan's first step anyway, and lets auto mode see how deep the tree is
if blocks.get(&root_cid).is_none() {
blocks.extend(
fetch_blocks(
http,
pds,
did,
&[root_cid],
&throttle,
&tier,
app_state.verify_cids,
app_state.max_car_body_bytes,
)
.await?,
);
if blocks.get(&root_cid).is_none() {
debug!(%root_cid, "pds did not serve the mst root");
return Ok(SparseBackfillResult::Skipped);
}
}
let auto_full_layer = if strategy == BackfillStrategy::Auto {
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, 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.get(cid).is_none()) {
return Ok(SparseBackfillResult::Skipped);
}
scanner.insert_blocks(blocks);
}
}
};
let scanned_blocks = scanner.take_blocks();
// leaves can share a cid (identical records), and each is fetched once
let missing_records = scan
.leaves
.iter()
.filter_map(|(_, cid)| scanned_blocks.get(cid).is_none().then_some(*cid))
.collect::<std::collections::BTreeSet<_>>()
.into_iter()
.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 fetched = fetch_blocks(
http,
pds,
did,
&missing_records,
&throttle,
&tier,
app_state.verify_cids,
app_state.max_car_body_bytes,
)
.await?;
if let Some(cid) = missing_records
.iter()
.find(|cid| fetched.get(cid).is_none())
{
return Err(miette::miette!("sparse record block missing after getBlocks: {cid}").into());
}
Ok(SparseBackfillResult::Fetched(Box::new(
SparseBackfillSuccess {
root_commit,
records: FilteredRecords::new(scan.leaves, scanned_blocks, fetched),
filter,
node_blocks: scan.node_blocks_seen,
node_bytes: scan.node_bytes_seen,
},
)))
}
/// sends one sparse request once the pds's throttle allows it. the response comes back as is,
/// so a caller can read a status that only means "no" before calling it a failure
async fn send_throttled(
http: &ThrottledHttpClient,
url: &url::Url,
accept: &'static str,
throttle: &ThrottleHandle,
tier: &RateTier,
) -> Result<http::Response<Vec<u8>>, BackfillError> {
let endpoint = PublicEndpoint::parse_http(url).map_err(|error| {
BackfillError::Generic(miette::miette!(
"unsafe public sparse endpoint {url}: {error}"
))
})?;
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::Request::builder()
.method(http::Method::GET)
.uri(endpoint.url().as_str())
.header(http::header::ACCEPT, accept)
.body(Vec::new())
.map_err(|e| BackfillError::Generic(miette::miette!("invalid sparse request: {e}")))?;
match http.send_http(request).await {
Ok(resp) => {
if (resp.status().is_success() || matches!(resp.status().as_u16(), 400 | 401))
&& !throttle.is_throttled()
{
throttle.record_success();
}
Ok(resp)
}
Err(e) => {
let reason = e.to_string();
if let Some(secs) = throttle.record_failure_detail("transport", reason.clone()) {
debug!(secs, %reason, "throttling pds after sparse transport error");
}
Err(BackfillError::Transport(reason.into()))
}
}
}
/// the repo's latest commit block on its own, for a pds that can't prove a record is absent.
/// the commit is trusted through its signature exactly like a proof's, so starting from it
/// costs a couple more requests but finds the same records. `None` sends the repo to a full
/// getRepo, which also records an inactive repo's status.
async fn latest_commit_start(
app_state: &AppState,
http: &ThrottledHttpClient,
did: &Did,
pds: &url::Url,
pds_endpoint: &PublicEndpoint,
throttle: &ThrottleHandle,
tier: &RateTier,
) -> Result<Option<CommitCar>, BackfillError> {
let mut latest_url = pds_endpoint
.url()
.join("xrpc/com.atproto.sync.getLatestCommit")
.map_err(|e| BackfillError::Generic(miette::miette!("Invalid URL: {e}")))?;
latest_url.query_pairs_mut().append_pair("did", did.as_str());
let resp = send_throttled(http, &latest_url, "application/json", throttle, tier).await?;
let status = resp.status();
let body = resp.into_body();
let latest: LatestCommitOutput = if status.is_success() {
serde_json::from_slice(&body).into_diagnostic()?
} else if status == StatusCode::NOT_FOUND {
return Ok(None);
} else if status == StatusCode::BAD_REQUEST
&& matches!(
xrpc_error_code(&body).as_deref(),
Some("RepoNotFound" | "RepoTakendown" | "RepoSuspended" | "RepoDeactivated")
)
{
return Ok(None);
} else if status == StatusCode::TOO_MANY_REQUESTS {
return Err(BackfillError::Ratelimited);
} else {
return Err(miette::miette!(
"getLatestCommit failed with HTTP {status}: {}",
String::from_utf8_lossy(&body)
)
.into());
};
let commit = latest.cid.to_ipld().into_diagnostic()?;
let blocks = fetch_blocks(
http,
pds,
did,
&[commit],
throttle,
tier,
app_state.verify_cids,
app_state.max_car_body_bytes,
)
.await?;
if blocks.get(&commit).is_none() {
debug!(%commit, "pds did not serve its latest commit");
return Ok(None);
}
Ok(Some(CommitCar {
root: commit,
blocks,
}))
}
/// a commit signed by this did's key could still name another did if two dids share a key,
/// and its records would then be stored under the wrong repo
pub(crate) fn check_commit_did(commit_did: &str, did: &Did) -> Result<(), BackfillError> {
if commit_did != did.as_str() {
return Err(miette::miette!("commit is for {commit_did}, not {did}").into());
}
Ok(())
}
pub(crate) 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<Blocks<CommitBlocks>, BackfillError> {
let mut out = Blocks::default();
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<Blocks<CommitBlocks>, 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 http.send_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"
)));
}
crate::car::parse_car_blocks(bytes::Bytes::from(body), verify_cids)
.map_err(BackfillError::from)
} 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 matches!(
xrpc_error_code(&body).as_deref(),
Some(
"BlockNotFound"
| "RepoNotFound"
| "RepoTakendown"
| "RepoSuspended"
| "RepoDeactivated"
)
) {
return Ok(Blocks::default());
}
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())
}
}
#[cfg(test)]
mod tests {
use super::*;
use crate::backfill::test_util::{public_client, serve, tangled_repo};
use crate::filter::FilterConfig;
use crate::mst::tests::node;
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 crate::car::compute_cid;
use std::collections::BTreeMap;
use std::sync::atomic::{AtomicUsize, Ordering};
use std::time::Duration;
use tokio::sync::Barrier;
use crate::backfill::worker::apply::Outcome as Backfilled;
use crate::backfill::worker::process::write_sparse;
use crate::db::types::DbRkey;
use crate::db::{self, keys};
use crate::filter::FilterMode;
use crate::types::RepoState;
/// how a sparse backfill in these tests ended
#[derive(Debug)]
enum Outcome {
/// its records went through the shard
Written,
/// the queue entry went away, or nothing matched the filter's signals
Dropped,
/// it fell back to a getRepo
Skipped,
}
#[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 =
crate::car::CarWriter::new(crate::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 address = serve(app).await;
let throttler = crate::util::throttle::Throttler::new(10);
let (pds, http) = public_client(address, &throttler);
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"
);
}
const GET_BLOCKS: &str = "com.atproto.sync.getBlocks";
const GET_RECORD: &str = "com.atproto.sync.getRecord";
/// serves each car for every call to its xrpc method.
async fn spawn_car_server(routes: Vec<(&str, bytes::Bytes)>) -> std::net::SocketAddr {
let app = routes
.into_iter()
.fold(Router::new(), |app, (method, car)| {
app.route(
&format!("/xrpc/{method}"),
get(move || {
let car = car.clone();
async move {
(
[(
reqwest::header::CONTENT_TYPE.as_str(),
"application/vnd.ipld.car",
)],
car,
)
}
}),
)
});
serve(app).await
}
async fn car_with_block(cid: IpldCid, payload: &[u8]) -> bytes::Bytes {
let mut car = Vec::new();
let mut writer =
crate::car::CarWriter::new(crate::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(
address: std::net::SocketAddr,
wanted: IpldCid,
verify_cids: bool,
max_body_bytes: usize,
) -> Result<Blocks<CommitBlocks>, BackfillError> {
let throttler = crate::util::throttle::Throttler::new(10);
let (pds, http) = public_client(address, &throttler);
let throttle = throttler.get_handle(&pds).await;
let did = Did::new_static("did:plc:aaaaaaaaaaaaaaaaaaaaaaaa").unwrap();
fetch_blocks(
&http,
&pds,
&did,
&[wanted],
&throttle,
&RateTier::trusted(),
verify_cids,
max_body_bytes,
)
.await
}
const PROOF_DID: &str = "did:plc:aaaaaaaaaaaaaaaaaaaaaaaa";
const PROOF_COLLECTION: &str = "sh.tangled.repo";
const PROOF_RKEY: &str = "3kzbif5moe22m";
/// a `getRecord` proof whose signed commit maps the record to `signed`. the mst node block
/// still claims that tree's cid but maps the record to `pointed`, and `served`, if any, is
/// the block sent under `pointed`'s cid. equal bodies give an honest proof.
async fn record_proof(signed: &[u8], pointed: &[u8], served: Option<&[u8]>) -> bytes::Bytes {
let key = format!("{PROOF_COLLECTION}/{PROOF_RKEY}");
let mst_root = |record: &[u8]| {
let entries = vec![(key.as_str(), compute_cid(record), None)];
serde_ipld_dagcbor::to_vec(&node(entries, None)).unwrap()
};
let signed_root = compute_cid(&mst_root(signed));
let commit = crate::mst::AtpCommit {
did: Did::new_static(PROOF_DID).unwrap(),
version: 3,
data: signed_root,
rev: "3l2abcdefgh22".parse().unwrap(),
prev: None,
sig: bytes::Bytes::new(),
}
.to_cbor()
.unwrap();
let commit_cid = compute_cid(&commit);
let mut car = Vec::new();
let header = crate::car::CarHeader::new_v1(vec![commit_cid]);
let mut writer = crate::car::CarWriter::new(header, &mut car);
let record = served.map(|served| (compute_cid(pointed), served.to_vec()));
let blocks = [(commit_cid, commit), (signed_root, mst_root(pointed))];
for (cid, bytes) in blocks.into_iter().chain(record) {
writer.write(cid, bytes).await.unwrap();
}
writer.finish().await.unwrap();
car.into()
}
/// runs a sparse backfill of `PROOF_DID` against a mock pds serving `routes`, and returns
/// the result with the record body hydrant stored, if any.
async fn backfill_from(
routes: Vec<(&str, bytes::Bytes)>,
verify_cids: bool,
) -> (Result<Outcome, BackfillError>, Option<fjall::Slice>) {
let address = spawn_car_server(routes).await;
backfill_at(address, verify_cids, BackfillStrategy::SparseFilter, true).await
}
/// `queued: false` runs it for a queue entry that's already gone
async fn backfill_at(
address: std::net::SocketAddr,
verify_cids: bool,
strategy: BackfillStrategy,
queued: bool,
) -> (Result<Outcome, BackfillError>, Option<fjall::Slice>) {
let (result, stored, _, _) =
backfill_buffering(address, verify_cids, strategy, queued, 1, &[]).await;
(result, stored)
}
/// [`backfill_at`] with the repo's metadata naming index `named` and live commits at
/// `buffered` revs waiting in its buffer, also returning the buffer keys left afterwards and
/// the car rev the metadata remembers
async fn backfill_buffering(
address: std::net::SocketAddr,
verify_cids: bool,
strategy: BackfillStrategy,
queued: bool,
named: u64,
buffered: &[&str],
) -> (
Result<Outcome, BackfillError>,
Option<fjall::Slice>,
Vec<Vec<u8>>,
Option<crate::db::types::DbTid>,
) {
let (result, _tmp, state) = run_sparse(
address,
verify_cids,
strategy,
queued,
named,
|state, batch| {
let did = Did::new_static(PROOF_DID).unwrap();
for rev in buffered {
let rev = crate::db::types::DbTid::from(
&rev.parse::<jacquard_common::types::tid::Tid>().unwrap(),
);
batch.insert(
&state.db.indexer.resync_buffer,
keys::resync_buffer_key(&did, rev),
b"buffered",
);
}
},
)
.await;
let did = Did::new_static(PROOF_DID).unwrap();
let stored = state
.db
.indexer
.head(keys::record_key(
&did,
PROOF_COLLECTION,
&DbRkey::new(PROOF_RKEY),
))
.unwrap();
let left = state
.db
.indexer
.resync_buffer
.prefix(keys::resync_buffer_prefix(&did))
.map(|guard| guard.key().unwrap().to_vec())
.collect();
let metadata = state
.db
.repo_metadata
.get(keys::repo_metadata_key(&did))
.unwrap()
.expect("repo metadata");
let backfilled_rev = db::deser_repo_meta(&metadata).unwrap().backfilled_rev;
(result, stored, left, backfilled_rev)
}
/// a sparse backfill of `PROOF_DID` for `sh.tangled.*`, through a shard of its own, over a
/// repo whose metadata names index `named` and that `seed` adds to. hands back the state it
/// left
async fn run_sparse(
address: std::net::SocketAddr,
verify_cids: bool,
strategy: BackfillStrategy,
queued: bool,
named: u64,
seed: impl FnOnce(&AppState, &mut fjall::OwnedWriteBatch),
) -> (
Result<Outcome, BackfillError>,
tempfile::TempDir,
Arc<AppState>,
) {
let tmp = tempfile::tempdir().unwrap();
let config = crate::config::Config {
database_path: tmp.path().to_path_buf(),
verify_cids,
..Default::default()
};
let state = Arc::new(AppState::new(&config).unwrap());
state.filter.store(Arc::new(filter(&["sh.tangled.*"], &[])));
let did = Did::new_static(PROOF_DID).unwrap();
let mut batch = state.db.inner.batch();
batch.insert(
&state.db.repos,
keys::repo_key(&did),
db::ser_repo_state(&RepoState::backfilling()).unwrap(),
);
batch.insert(
&state.db.repo_metadata,
keys::repo_metadata_key(&did),
db::ser_repo_meta(&crate::types::RepoMetadata::backfilling(named)).unwrap(),
);
let pending_key = keys::pending_key(1);
if queued {
state
.db
.indexer
.stage_pending_insert(&mut batch, &did, &pending_key);
}
seed(&state, &mut batch);
batch.commit().unwrap();
let (pds, http) = public_client(address, &state.throttler);
let _shard = crate::ingest::indexer::FirehoseWorker::serve_one_shard(state.clone());
let result = match process_did_sparse(&state, &http, &did, &pds, false, strategy).await {
Ok(SparseBackfillResult::Fetched(sparse)) => {
let indexer = state.indexer.lock().clone().expect("a running shard");
write_sparse(
&state,
&indexer,
&did,
&pending_key.as_slice().into(),
RepoState::backfilling(),
RepoState::backfilling(),
*sparse,
)
.await
.map(|backfilled| match backfilled {
Backfilled::Drained | Backfilled::Skipped | Backfilled::Stopping => {
Outcome::Written
}
Backfilled::Superseded | Backfilled::Discarded | Backfilled::Inactive => {
Outcome::Dropped
}
})
}
Ok(SparseBackfillResult::Skipped) => Ok(Outcome::Skipped),
Err(e) => Err(e),
};
(result, tmp, state)
}
const LATEST_REV: &str = "3l2abcdefgh22";
const OTHER_DID: &str = "did:plc:bbbbbbbbbbbbbbbbbbbbbbbb";
/// one record under `PROOF_COLLECTION/PROOF_RKEY`, its mst root and a commit naming
/// `commit_did`, by cid
struct TestRepo {
commit: IpldCid,
blocks: BTreeMap<IpldCid, Vec<u8>>,
}
fn test_repo(commit_did: &'static str, record: &[u8]) -> TestRepo {
let key = format!("{PROOF_COLLECTION}/{PROOF_RKEY}");
let record_cid = compute_cid(record);
let root = serde_ipld_dagcbor::to_vec(&node(vec![(key.as_str(), record_cid, None)], None))
.unwrap();
let root_cid = compute_cid(&root);
let commit = crate::mst::AtpCommit {
did: Did::new_static(commit_did).unwrap(),
version: 3,
data: root_cid,
rev: LATEST_REV.parse().unwrap(),
prev: None,
sig: bytes::Bytes::new(),
}
.to_cbor()
.unwrap();
let commit_cid = compute_cid(&commit);
TestRepo {
commit: commit_cid,
blocks: BTreeMap::from([
(commit_cid, commit.to_vec()),
(root_cid, root),
(record_cid, record.to_vec()),
]),
}
}
type Reply = (StatusCode, &'static str, String);
fn json_error(status: StatusCode, error: &str) -> Reply {
let body = serde_json::json!({ "error": error, "message": "nope" }).to_string();
(status, "application/json", body)
}
fn reply_route(reply: Reply) -> axum::routing::MethodRouter {
get(move || {
let (status, content_type, body) = reply.clone();
async move {
(
status,
[(reqwest::header::CONTENT_TYPE.as_str(), content_type)],
body,
)
}
})
}
/// a pds without exclusion proofs: the getRecord probe gets `probe`, getLatestCommit gets
/// `latest` (pointing at `repo`'s commit when `None`), and getBlocks serves only the blocks
/// asked for, from `blocks`
fn no_proof_pds(
repo: &TestRepo,
probe: Reply,
latest: Option<Reply>,
blocks: Option<BTreeMap<IpldCid, Vec<u8>>>,
) -> Router {
let latest = latest.unwrap_or_else(|| {
let body = serde_json::json!({ "cid": repo.commit.to_string(), "rev": LATEST_REV });
(StatusCode::OK, "application/json", body.to_string())
});
let app = Router::new()
.route(&format!("/xrpc/{GET_RECORD}"), reply_route(probe))
.route(
"/xrpc/com.atproto.sync.getLatestCommit",
reply_route(latest),
);
let Some(blocks) = blocks else {
return app;
};
let blocks = Arc::new(blocks);
app.route(
&format!("/xrpc/{GET_BLOCKS}"),
get(
move |axum::extract::RawQuery(query): axum::extract::RawQuery| {
let blocks = blocks.clone();
async move {
let found =
url::form_urlencoded::parse(query.unwrap_or_default().as_bytes())
.filter(|(name, _)| name == "cids")
.filter_map(|(_, cid)| cid.parse::<IpldCid>().ok())
.filter_map(|cid| {
blocks.get(&cid).map(|bytes| (cid, bytes.clone()))
})
.collect::<Vec<_>>();
if found.is_empty() {
let (status, content_type, body) =
json_error(StatusCode::BAD_REQUEST, "BlockNotFound");
let headers = [(reqwest::header::CONTENT_TYPE.as_str(), content_type)];
return (status, headers, body.into_bytes());
}
let mut car = Vec::new();
let header = crate::car::CarHeader::new_v1(Vec::new());
let mut writer = crate::car::CarWriter::new(header, &mut car);
for (cid, bytes) in found {
writer.write(cid, bytes).await.unwrap();
}
writer.finish().await.unwrap();
let headers = [(
reqwest::header::CONTENT_TYPE.as_str(),
"application/vnd.ipld.car",
)];
(StatusCode::OK, headers, car)
}
},
),
)
}
async fn backfill_no_proof(
app: Router,
strategy: BackfillStrategy,
) -> (Result<Outcome, BackfillError>, Option<fjall::Slice>) {
backfill_at(serve(app).await, true, strategy, true).await
}
#[tokio::test]
async fn a_backfill_whose_queue_entry_went_away_writes_nothing() {
let repo = test_repo(PROOF_DID, &tangled_repo("honest"));
let probe = json_error(StatusCode::NOT_FOUND, "RecordNotFound");
let app = no_proof_pds(&repo, probe, None, Some(repo.blocks.clone()));
let (result, stored) = backfill_at(
serve(app).await,
true,
BackfillStrategy::SparseFilter,
false,
)
.await;
assert!(matches!(result, Ok(Outcome::Dropped)), "{result:?}");
assert_eq!(stored, None);
}
#[tokio::test]
async fn a_sparse_backfill_drops_the_buffered_commits_its_commit_covers_and_keeps_its_rev() {
let repo = test_repo(PROOF_DID, &tangled_repo("honest"));
let probe = json_error(StatusCode::NOT_FOUND, "RecordNotFound");
let app = no_proof_pds(&repo, probe, None, Some(repo.blocks.clone()));
// a newer commit would be applied by the drain right after, which the task's tests
// cover with real commits
let (result, _, left, backfilled_rev) = backfill_buffering(
serve(app).await,
true,
BackfillStrategy::SparseFilter,
true,
1,
&["3l2abcdefgg22", LATEST_REV],
)
.await;
assert!(matches!(result, Ok(Outcome::Written)), "{result:?}");
assert!(left.is_empty(), "{left:?}");
// so a commit at or below it that the shard gets after this doesn't apply either
assert_eq!(
backfilled_rev,
Some(crate::db::types::DbTid::from(
&LATEST_REV
.parse::<jacquard_common::types::tid::Tid>()
.unwrap(),
))
);
}
/// a sparse backfill over a repo that already stores a post, which the filter doesn't
/// fetch, and a tangled repo record the fetch won't find again
async fn sparse_over_stored_records() -> (
Result<Outcome, BackfillError>,
tempfile::TempDir,
Arc<AppState>,
) {
let repo = test_repo(PROOF_DID, &tangled_repo("honest"));
let probe = json_error(StatusCode::NOT_FOUND, "RecordNotFound");
let app = no_proof_pds(&repo, probe, None, Some(repo.blocks.clone()));
run_sparse(
serve(app).await,
true,
BackfillStrategy::SparseFilter,
true,
1,
|state, batch| {
let did = Did::new_static(PROOF_DID).unwrap();
for (collection, rkey, count) in [
("app.bsky.feed.post", "3kzbif5moe2aa", 5),
(PROOF_COLLECTION, "3kzbif5moe2zz", 9),
] {
let key = keys::record_key(&did, collection, &DbRkey::new(rkey));
state.db.indexer.stage_head(batch, key, tangled_repo(rkey));
db::set_record_count(batch, &state.db, &did, collection, count);
}
},
)
.await
}
#[tokio::test]
async fn a_sparse_backfill_leaves_the_records_of_collections_it_didnt_fetch() {
let (result, _tmp, state) = sparse_over_stored_records().await;
assert!(matches!(result, Ok(Outcome::Written)), "{result:?}");
let did = Did::new_static(PROOF_DID).unwrap();
let stored = |collection, rkey| {
state
.db
.indexer
.head(keys::record_key(&did, collection, &DbRkey::new(rkey)))
.unwrap()
};
assert!(stored("app.bsky.feed.post", "3kzbif5moe2aa").is_some());
assert!(stored(PROOF_COLLECTION, PROOF_RKEY).is_some());
// a record of a fetched collection that the fetch didn't find again is gone
assert!(stored(PROOF_COLLECTION, "3kzbif5moe2zz").is_none());
}
#[tokio::test]
async fn a_sparse_backfill_replaces_only_the_counts_of_collections_it_fetched() {
let (result, _tmp, state) = sparse_over_stored_records().await;
assert!(matches!(result, Ok(Outcome::Written)), "{result:?}");
let did = Did::new_static(PROOF_DID).unwrap();
let count = |collection| db::get_record_count(&state.db, &did, collection).unwrap();
assert_eq!(count("app.bsky.feed.post"), 5);
assert_eq!(count(PROOF_COLLECTION), 1);
}
#[tokio::test]
async fn a_backfill_for_a_key_the_metadata_doesnt_name_writes_nothing() {
let repo = test_repo(PROOF_DID, &tangled_repo("honest"));
let probe = json_error(StatusCode::NOT_FOUND, "RecordNotFound");
let app = no_proof_pds(&repo, probe, None, Some(repo.blocks.clone()));
// queued under 1 while the metadata names 2, so 1 is an orphan
let (result, stored, _, _) = backfill_buffering(
serve(app).await,
true,
BackfillStrategy::SparseFilter,
true,
2,
&[],
)
.await;
assert!(matches!(result, Ok(Outcome::Dropped)), "{result:?}");
assert_eq!(stored, None);
}
#[tokio::test]
async fn probe_without_proof_starts_from_the_latest_commit() {
let honest = tangled_repo("honest");
let repo = test_repo(PROOF_DID, &honest);
// a pds that can't prove absence (an atcr hold's text 404), tranquil's 404, and the
// 400 an xrpc client expects
let probes = [
(
StatusCode::NOT_FOUND,
"text/plain",
"mst: not found".to_string(),
),
json_error(StatusCode::NOT_FOUND, "RecordNotFound"),
json_error(StatusCode::BAD_REQUEST, "RecordNotFound"),
];
for probe in probes {
let app = no_proof_pds(&repo, probe.clone(), None, Some(repo.blocks.clone()));
let (result, stored) = backfill_no_proof(app, BackfillStrategy::SparseFilter).await;
assert!(
matches!(result, Ok(Outcome::Written)),
"probe {probe:?}: expected a write, got {result:?}"
);
assert_eq!(
stored.as_deref(),
Some(honest.as_slice()),
"probe {probe:?}"
);
}
}
#[tokio::test]
async fn latest_commit_for_another_did_is_rejected() {
let repo = test_repo(OTHER_DID, &tangled_repo("honest"));
let probe = json_error(StatusCode::NOT_FOUND, "RecordNotFound");
let app = no_proof_pds(&repo, probe, None, Some(repo.blocks.clone()));
let (result, stored) = backfill_no_proof(app, BackfillStrategy::SparseFilter).await;
let err = result.unwrap_err();
assert!(err.to_string().contains("commit is for"), "{err}");
assert_eq!(stored, None);
}
#[tokio::test]
async fn proof_commit_for_another_did_is_rejected() {
let repo = test_repo(OTHER_DID, &tangled_repo("honest"));
let mut car = Vec::new();
let mut writer =
crate::car::CarWriter::new(crate::car::CarHeader::new_v1(vec![repo.commit]), &mut car);
for (cid, bytes) in &repo.blocks {
writer.write(*cid, bytes.clone()).await.unwrap();
}
writer.finish().await.unwrap();
let (result, stored) = backfill_from(vec![(GET_RECORD, car.into())], true).await;
let err = result.unwrap_err();
assert!(err.to_string().contains("commit is for"), "{err}");
assert_eq!(stored, None);
}
#[tokio::test]
async fn forged_latest_commit_block_is_rejected() {
// getLatestCommit names one commit, and getBlocks serves another under its cid
let honest = tangled_repo("honest");
let repo = test_repo(PROOF_DID, &honest);
let forged = test_repo(PROOF_DID, &tangled_repo("forged"));
let mut blocks = forged.blocks.clone();
blocks.insert(repo.commit, forged.blocks[&forged.commit].clone());
let probe = json_error(StatusCode::NOT_FOUND, "RecordNotFound");
let app = no_proof_pds(&repo, probe, None, Some(blocks));
let (result, stored) = backfill_no_proof(app, BackfillStrategy::SparseFilter).await;
let err = result.unwrap_err();
assert!(err.to_string().contains("CAR block CID mismatch"), "{err}");
assert_eq!(stored, None);
}
#[tokio::test]
async fn latest_commit_that_ends_the_sparse_attempt_skips_to_full() {
let repo = test_repo(PROOF_DID, &tangled_repo("honest"));
let probe = json_error(StatusCode::NOT_FOUND, "RecordNotFound");
let cases = [
(
"inactive repo",
Some(json_error(StatusCode::BAD_REQUEST, "RepoDeactivated")),
Some(repo.blocks.clone()),
),
(
"no getLatestCommit",
Some((StatusCode::NOT_FOUND, "text/plain", String::new())),
Some(repo.blocks.clone()),
),
("commit not served", None, Some(BTreeMap::new())),
];
for (case, latest, blocks) in cases {
let app = no_proof_pds(&repo, probe.clone(), latest, blocks);
let (result, stored) = backfill_no_proof(app, BackfillStrategy::SparseFilter).await;
assert!(
matches!(result, Ok(Outcome::Skipped)),
"{case}: expected a skip, got {result:?}"
);
assert_eq!(stored, None, "{case}");
}
}
#[tokio::test]
async fn pds_without_get_blocks_fails_the_sparse_attempt() {
let repo = test_repo(PROOF_DID, &tangled_repo("honest"));
let probe = (
StatusCode::NOT_FOUND,
"text/plain",
"mst: not found".to_string(),
);
let app = no_proof_pds(&repo, probe, None, None);
let (result, stored) = backfill_no_proof(app, BackfillStrategy::SparseFilter).await;
let err = result.unwrap_err();
assert!(
err.to_string().contains("getBlocks failed with HTTP 404"),
"{err}"
);
assert_eq!(stored, None);
}
#[tokio::test]
async fn auto_still_takes_full_for_a_tiny_tree_reached_from_the_latest_commit() {
let repo = test_repo(PROOF_DID, &tangled_repo("honest"));
let probe = json_error(StatusCode::NOT_FOUND, "RecordNotFound");
let app = no_proof_pds(&repo, probe, None, Some(repo.blocks.clone()));
let (result, stored) = backfill_no_proof(app, BackfillStrategy::Auto).await;
assert!(
matches!(result, Ok(Outcome::Skipped)),
"expected auto to pick full, got {result:?}"
);
assert_eq!(stored, None);
}
fn assert_imported_one(result: Result<Outcome, BackfillError>) {
assert!(matches!(result, Ok(Outcome::Written)), "{result:?}");
}
#[tokio::test]
async fn verify_cids_accepts_honest_get_record_proof() {
let honest = tangled_repo("honest");
let proof = record_proof(&honest, &honest, Some(&honest)).await;
let (result, stored) = backfill_from(vec![(GET_RECORD, proof)], true).await;
assert_imported_one(result);
assert_eq!(stored.as_deref(), Some(honest.as_slice()));
}
#[tokio::test]
async fn verify_cids_rejects_forged_mst_node_in_get_record_proof() {
let (honest, forged) = (tangled_repo("honest"), tangled_repo("forged"));
let proof = record_proof(&honest, &forged, Some(&forged)).await;
let (result, stored) = backfill_from(vec![(GET_RECORD, proof)], true).await;
let err = result.unwrap_err();
assert!(err.to_string().contains("CAR block CID mismatch"), "{err}");
assert_eq!(stored, None);
}
#[tokio::test]
async fn verify_cids_disabled_imports_forged_mst_node_from_get_record_proof() {
// the forged node's leaf and the record it points at agree, so only the cid check can
// tell that the signed tree never contained this record
let (honest, forged) = (tangled_repo("honest"), tangled_repo("forged"));
let proof = record_proof(&honest, &forged, Some(&forged)).await;
let (result, stored) = backfill_from(vec![(GET_RECORD, proof)], false).await;
assert_imported_one(result);
assert_eq!(stored.as_deref(), Some(forged.as_slice()));
}
#[tokio::test]
async fn forged_record_body_is_rejected_without_verify_cids() {
// a block the parse did not check is hashed before it is stored
let (honest, forged) = (tangled_repo("honest"), tangled_repo("forged"));
let proof = record_proof(&honest, &honest, Some(&forged)).await;
let (result, stored) = backfill_from(vec![(GET_RECORD, proof)], false).await;
let err = result.unwrap_err();
assert!(err.to_string().contains("CAR block CID mismatch"), "{err}");
assert!(err.to_string().contains(PROOF_COLLECTION), "{err}");
assert_eq!(stored, None);
}
#[tokio::test]
async fn record_missing_from_proof_is_fetched_with_get_blocks() {
let honest = tangled_repo("honest");
let proof = record_proof(&honest, &honest, None).await;
let record = car_with_block(compute_cid(&honest), &honest).await;
let (result, stored) =
backfill_from(vec![(GET_RECORD, proof), (GET_BLOCKS, record)], true).await;
assert_imported_one(result);
assert_eq!(stored.as_deref(), Some(honest.as_slice()));
}
#[tokio::test]
async fn forged_record_from_get_blocks_is_rejected_without_verify_cids() {
let (honest, forged) = (tangled_repo("honest"), tangled_repo("forged"));
let proof = record_proof(&honest, &honest, None).await;
let record = car_with_block(compute_cid(&honest), &forged).await;
let (result, stored) =
backfill_from(vec![(GET_RECORD, proof), (GET_BLOCKS, record)], false).await;
let err = result.unwrap_err();
assert!(err.to_string().contains("CAR block CID mismatch"), "{err}");
assert_eq!(stored, None);
}
#[tokio::test]
async fn verify_cids_rejects_mismatched_get_blocks_car() {
let pds =
spawn_car_server(vec![(GET_BLOCKS, 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 = compute_cid(b"trusted");
let pds = spawn_car_server(vec![(
GET_BLOCKS,
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(vec![(GET_BLOCKS, 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(vec![(
GET_BLOCKS,
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"));
}
}