use std::collections::HashSet;
use std::sync::Arc;
use std::time::Instant;
use bytes::Bytes;
use fjall::Slice;
use miette::{IntoDiagnostic, Result};
use reqwest::StatusCode;
use tracing::{debug, trace, warn};
use crate::backfill::sparse::xrpc_error_code;
use jacquard_common::IntoStatic;
use jacquard_common::types::cid::IpldCid;
use jacquard_common::types::did::Did;
use crate::backfill::admission::BackfillAdmission;
use crate::backfill::budget::{BackfillBudget, Reservation, peak_for, spilled_peak_for};
use crate::backfill::client::{ThrottledHttpClient, collect_body_bounded};
use crate::backfill::error::BackfillError;
use crate::backfill::filtered_car::{FilteredCar, FilteredRepo};
use crate::backfill::sparse::{
SPARSE_MAX_SCAN_ROUNDS, SparseBackfillResult, SparseBackfillSuccess, check_commit_did,
fetch_blocks, process_did_sparse,
};
use crate::backfill::spill::{SpillFile, SpillingCar, Spills};
use crate::backfill::worker::apply::{self, Outcome};
use crate::backfill::worker::records::{CheckOutcome, FilteredRecords, RepoRecords, WholeBlocks, check_records};
use crate::car::{Blocks, ParsedCar, WholeCar};
use crate::config::{BackfillStrategy, RateTier};
use crate::db::{Txn as DbTxn, keys};
use crate::filter::FilterConfig;
use crate::ingest::indexer::IndexerTx;
use crate::ingest::indexer::message::BackfillTail;
use crate::ingest::indexer::repos::write_repo;
use crate::mst::BlockSource;
use crate::net::PublicEndpoint;
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::db::outbox::StreamOutbox;
#[cfg(feature = "indexer_stream")]
use crate::types::{AccountEvt, IdentityEvt};
/// 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(crate) struct BackfillImported {
pub(crate) before: RepoState<'static>,
pub(crate) after: RepoState<'static>,
}
impl BackfillImported {
/// the account and identity events for what the backfill changed about the repo
#[cfg(feature = "indexer_stream")]
pub(crate) fn stage(self, did: &Did, events: &mut StreamOutbox) {
if self.after.account_moved_from(&self.before) {
events.push_account(account_event(did, &self.after.status, self.after.active));
}
if self.after.identity_moved_from(&self.before) {
events.push_identity(IdentityEvt {
did: did.clone().into_static(),
handle: self.after.handle,
});
}
}
}
#[cfg(feature = "indexer_stream")]
fn account_event(did: &Did, status: &RepoStatus, active: bool) -> AccountEvt<'static> {
let status = match status {
RepoStatus::Deactivated => "deactivated",
RepoStatus::Takendown => "takendown",
RepoStatus::Suspended => "suspended",
RepoStatus::Deleted => "deleted",
RepoStatus::Desynchronized => "desynchronized",
RepoStatus::Throttled => "throttled",
_ => "",
};
AccountEvt {
did: did.clone().into_static(),
active,
status: (!status.is_empty()).then(|| status.into()),
}
}
#[allow(clippy::too_many_arguments)]
pub(super) async fn process_did(
app_state: &Arc<AppState>,
http: &ThrottledHttpClient,
indexer: &IndexerTx,
did: &Did,
pending_key: Slice,
admission: &BackfillAdmission,
verify_signatures: bool,
strategy: BackfillStrategy,
) -> Result<Outcome, 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"
);
// the doc was invalidated above, so this is a fresh fetch
state.update_from_resolve(doc, chrono::Utc::now().timestamp());
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, verify_signatures, strategy).await
{
Ok(SparseBackfillResult::Fetched(sparse)) => {
trace!(
node_blocks = sparse.node_blocks,
node_bytes = sparse.node_bytes,
"sparse fetch complete"
);
// the pds is done with, and the steps can wait a while on a busy shard
drop(admission_permit);
return write_sparse(
app_state,
indexer,
did,
&pending_key,
previous_state,
state,
*sparse,
)
.await;
}
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). one filter snapshot serves both the fetch and the insert below, so a
// filter change in between can't delete stored records the fetch never kept
let start = Instant::now();
let filter = app_state.filter.load_full();
let throttle = app_state.throttler.get_handle(&pds).await;
let tier = app_state.resolve_pds_tier(pds.host_str().unwrap_or(""));
let car = FilteredCar::new(filter.clone(), app_state.verify_cids).map_or_else(
|| {
RepoCar::whole(
app_state.verify_cids,
app_state.backfill_budget.clone(),
app_state.spills.clone(),
)
},
RepoCar::Filtered,
);
// a whole car's budget is held until its last step is in, since every block stays until then
let (car, _budget) = match fetch_full_repo_car(
http,
&pds,
did,
&throttle,
&tier,
app_state.max_car_body_bytes,
car,
)
.await?
{
FullRepoOutcome::Car { car, budget } => (ParsedRepo::Whole(car), budget),
FullRepoOutcome::Filtered(repo) => (ParsedRepo::Filtered(repo), None),
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 account = account_event(did, &status, false);
let resync_state = ResyncState::Gone {
status: status.clone(),
};
let resync_bytes = rmp_serde::to_vec(&resync_state).into_diagnostic()?;
let owner = did.clone();
let did = did.clone();
let pending_key = pending_key.clone();
write_repo(app_state, &owner, move |db| {
let mut txn = DbTxn::new(db);
if txn.is_excluded(&did)? {
return Ok(());
}
let applied = txn.transition_pending_key(
&did,
pending_key.as_ref(),
GaugeState::Resync(None),
)?;
db.indexer
.stage_pending_remove(&mut txn.batch, &did, &pending_key);
if applied {
if let Some(mut state) = txn.read_repo(&did)? {
state.active = false;
state.status = status;
txn.batch
.insert(&db.indexer.resync, keys::repo_key(&did), resync_bytes);
txn.put_repo(&did, &state)?;
}
#[cfg(feature = "indexer_stream")]
txn.outbox.stream.push_account(account);
}
txn.commit()
})
.await?;
return Ok(Outcome::Inactive);
}
};
// a filtered car can still go back to the pds for records it dropped, so it keeps
// its admission through the mst walk
let admission_permit = match &car {
ParsedRepo::Whole(_) => {
drop(admission_permit);
None
}
ParsedRepo::Filtered(_) => Some(admission_permit),
};
match &car {
ParsedRepo::Whole(repo) => trace!(
blocks = repo.blocks.len(),
spilled = matches!(repo.blocks, WholeBlocks::Spilled(_)),
elapsed = ?start.elapsed(),
"streamed whole car"
),
ParsedRepo::Filtered(repo) => {
let stats = repo.stats;
trace!(
bytes = stats.bytes,
blocks = stats.blocks,
kept_bytes = stats.kept_bytes,
kept_blocks = stats.kept_blocks,
ruled_out = repo.scanner.ruled_out_len(),
elapsed = ?start.elapsed(),
"streamed car"
);
}
}
// 4. parse root commit to get mst root
let root_bytes = match &car {
ParsedRepo::Whole(repo) => repo.blocks.read(&repo.root)?,
ParsedRepo::Filtered(repo) => repo.blocks.get(&repo.root).cloned(),
}
.ok_or_else(|| miette::miette!("root block missing from CAR"))?;
let root_commit = crate::mst::AtpCommit::from_cbor(&root_bytes)?;
check_commit_did(root_commit.did.as_str(), did)?;
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. find every record block
let start = Instant::now();
let mst_root = root_commit.data;
let records = match car {
ParsedRepo::Whole(repo) => RepoRecords::Whole {
blocks: repo.blocks,
mst_root,
},
ParsedRepo::Filtered(repo) => RepoRecords::Filtered(
filtered_records(
app_state, http, did, &pds, &throttle, &tier, *repo, mst_root,
)
.await?,
),
};
drop(admission_permit);
trace!(elapsed = %start.elapsed().as_secs_f32(), "found records");
write_records(
app_state,
indexer,
did,
&pending_key,
Fetch {
before: previous_state,
after: state,
root_commit,
records,
filter,
fetched_only: None,
},
)
.await
}
/// writes what a sparse fetch found the way [`write_records`] writes a getRepo's, except that
/// it only touches the collections the fetch asked for
pub(crate) async fn write_sparse(
app_state: &Arc<AppState>,
indexer: &IndexerTx,
did: &Did,
pending_key: &Slice,
before: RepoState<'static>,
after: RepoState<'static>,
sparse: SparseBackfillSuccess,
) -> Result<Outcome, BackfillError> {
let SparseBackfillSuccess {
root_commit,
records,
filter,
..
} = sparse;
let fetch = Fetch {
before,
after,
root_commit,
records: RepoRecords::Filtered(records),
fetched_only: Some(filter.clone()),
filter,
};
write_records(app_state, indexer, did, pending_key, fetch).await
}
/// a fetched repo, ready to write
struct Fetch {
/// the row as the backfill read it before the fetch
before: RepoState<'static>,
/// `before` with the doc resolved before the fetch
after: RepoState<'static>,
root_commit: Commit,
records: RepoRecords,
/// the filter the fetch kept records for
filter: Arc<FilterConfig>,
/// see [`BackfillTail::fetched_only`]
fetched_only: Option<Arc<FilterConfig>>,
}
/// writes the records through the repo's indexer shard, a step at a time. everything but the
/// writes themselves runs here, off the shard
async fn write_records(
app_state: &Arc<AppState>,
indexer: &IndexerTx,
did: &Did,
pending_key: &Slice,
fetch: Fetch,
) -> Result<Outcome, BackfillError> {
let Fetch {
before,
mut after,
root_commit,
records,
filter,
fetched_only,
} = fetch;
let start = Instant::now();
let rev = root_commit.rev;
after.root = Some(root_commit);
after.touch();
let outcome = {
let app_state = app_state.clone();
let indexer = indexer.clone();
let did = did.clone().into_static();
let pending_key = pending_key.clone();
tokio::task::spawn_blocking(move || {
// a first pass over every record before anything is sent, so a broken mst or a key
// that doesn't parse fails with nothing written
let mut checked = match check_records(records, filter)? {
CheckOutcome::Keep(checked) => checked,
CheckOutcome::Discarded => {
trace!("no signal-matching records found, discarding repo");
return Ok(Outcome::Discarded);
}
};
if checked.left_out > 0 {
debug!(
records = checked.left_out,
"skipping records whose block the car left out"
);
}
let tail = BackfillTail {
before,
after,
collection_counts: checked.take_collection_counts(),
fetched_only,
};
apply::write(&app_state, &indexer, &did, &pending_key, rev, &checked, tail)
})
.await
.into_diagnostic()??
};
trace!(elapsed = %start.elapsed().as_secs_f32(), "wrote records");
match outcome {
// signal mode adds the repo optimistically, so drop it when nothing matched
Outcome::Discarded => remove_discarded_repo(app_state, did, pending_key).await?,
Outcome::Superseded => {
debug!("queue entry went away during the fetch, dropping what it fetched")
}
_ => {}
}
Ok(outcome)
}
async fn remove_discarded_repo(
app_state: &Arc<AppState>,
did: &Did,
pending_key: &Slice,
) -> Result<(), BackfillError> {
let metadata_key = keys::repo_metadata_key(did);
let owner = did.clone();
let did = did.clone();
let pending_key = pending_key.clone();
write_repo(app_state, &owner, move |db| {
let mut txn = DbTxn::new(db);
if txn.is_excluded(&did)? {
return Ok(());
}
let applied = txn.transition_pending_key(&did, pending_key.as_ref(), GaugeState::Synced)?;
db.indexer
.stage_pending_remove(&mut txn.batch, &did, &pending_key);
if applied {
txn.remove_repo(&did)?;
txn.batch.remove(&db.repo_metadata, &metadata_key);
txn.counts.add_repos(-1);
}
txn.commit()
})
.await?;
Ok(())
}
/// a getRepo body, ready for the mst walk
enum ParsedRepo {
Whole(WholeRepo),
Filtered(Box<FilteredRepo>),
}
/// a whole getRepo body's root commit and its blocks
pub(super) struct WholeRepo {
root: IpldCid,
blocks: WholeBlocks,
}
impl std::fmt::Debug for WholeRepo {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.debug_struct("WholeRepo")
.field("root", &self.root)
.field("blocks", &self.blocks.len())
.finish()
}
}
/// what a getRepo body is read into as it streams
enum RepoCar {
Whole {
body: WholeBody,
budget: BackfillBudget,
spills: Spills,
},
Filtered(FilteredCar),
}
impl RepoCar {
fn whole(verify_cids: bool, budget: BackfillBudget, spills: Spills) -> Self {
RepoCar::Whole {
body: WholeBody::Memory(WholeCar::new(verify_cids)),
budget,
spills,
}
}
fn push(&mut self, chunk: &[u8]) -> Result<()> {
match self {
RepoCar::Whole { body, .. } => body.push(chunk),
RepoCar::Filtered(car) => car.push(chunk),
}
}
/// carries on writing the car to `file`, with what it read so far
fn spill(self, file: SpillFile) -> Result<Self> {
Ok(match self {
RepoCar::Whole {
body,
budget,
spills,
} => RepoCar::Whole {
body: body.spill(file)?,
budget,
spills,
},
filtered @ RepoCar::Filtered(_) => filtered,
})
}
fn finish(self) -> Result<FullRepoOutcome> {
Ok(match self {
RepoCar::Whole { body, .. } => FullRepoOutcome::Car {
car: body.finish()?,
budget: None,
},
RepoCar::Filtered(car) => FullRepoOutcome::Filtered(Box::new(car.finish()?)),
})
}
}
/// a whole car as it streams, held in memory until it gets big enough to spill
enum WholeBody {
Memory(WholeCar),
Spilling(SpillingCar),
}
impl WholeBody {
fn push(&mut self, chunk: &[u8]) -> Result<()> {
match self {
WholeBody::Memory(car) => car.push(chunk),
WholeBody::Spilling(car) => car.push(chunk),
}
}
fn spill(self, file: SpillFile) -> Result<Self> {
Ok(match self {
WholeBody::Memory(car) => WholeBody::Spilling(SpillingCar::from_memory(car, file)?),
spilling @ WholeBody::Spilling(_) => spilling,
})
}
fn finish(self) -> Result<WholeRepo> {
Ok(match self {
WholeBody::Memory(car) => {
let ParsedCar { root, blocks } = car.finish()?;
WholeRepo {
root,
blocks: WholeBlocks::Memory(blocks),
}
}
WholeBody::Spilling(car) => {
let (root, spilled) = car.finish()?;
WholeRepo {
root,
blocks: WholeBlocks::Spilled(spilled),
}
}
})
}
}
/// what goes from the download to the thread reading the car
enum Read {
Chunk(Bytes),
/// the car got too big to hold, so it goes to this file from here on
Spill(SpillFile),
}
/// a whole car's share of the backfill budget as it downloads, and whether it spilled
struct WholeFetch {
reservation: Reservation,
spills: Spills,
spilled: bool,
}
impl WholeFetch {
/// counts `received` bytes of the car against the budget, and spills it once it passes
/// the threshold. once it holds a spill slot it never waits on the budget, see
/// [`Reservation::resize_now`]
async fn received(
&mut self,
received: usize,
reader: &tokio::sync::mpsc::Sender<Read>,
) -> Result<(), BackfillError> {
if !self.spilled && received > self.spills.threshold() {
let file = self.spills.start().await?;
// a reader that already stopped reports why when it's joined
let _ = reader.send(Read::Spill(file)).await;
self.spilled = true;
}
if self.spilled {
self.reservation.resize_now(spilled_peak_for(received));
} else {
self.reservation.grow_to(peak_for(received)).await;
}
Ok(())
}
}
/// how many chunks the car reader can fall behind the download before the download waits
const CHUNKS_IN_FLIGHT: usize = 16;
/// feeds a getRepo body to `car` on a blocking thread, since hashing and decoding every block
/// of a big repo would otherwise hold up the runtime. `None` once the body passes `max_bytes`,
/// counted on decoded bytes like [`collect_body_bounded`]
async fn stream_car(
mut resp: reqwest::Response,
max_bytes: usize,
mut car: RepoCar,
) -> Result<Option<FullRepoOutcome>, BackfillError> {
let content_length = resp.content_length();
if content_length.is_some_and(|len| len > max_bytes as u64) {
return Ok(None);
}
// taken after the backfill permit, the pds admission and the pds connection permit and
// never before them, so a task waiting here holds nothing a budget holder could need
let mut whole = match &car {
RepoCar::Whole { budget, spills, .. } => Some(WholeFetch {
reservation: budget.reservation(),
spills: spills.clone(),
spilled: false,
}),
RepoCar::Filtered(_) => None,
};
if let Some(fetch) = &mut whole {
let len = content_length.map(|len| len as usize);
let threshold = fetch.spills.threshold();
if threshold == 0 || len.is_some_and(|len| len > threshold) {
car = car.spill(fetch.spills.start().await?)?;
fetch.spilled = true;
fetch
.reservation
.resize_now(spilled_peak_for(len.unwrap_or(0)));
} else if let Some(len) = len {
fetch.reservation.grow_to(peak_for(len)).await;
if let RepoCar::Whole {
body: WholeBody::Memory(body),
..
} = &mut car
{
body.reserve(len);
}
}
}
let (tx, mut rx) = tokio::sync::mpsc::channel::<Read>(CHUNKS_IN_FLIGHT);
let reader = tokio::task::spawn_blocking(move || {
while let Some(read) = rx.blocking_recv() {
match read {
Read::Chunk(chunk) => car.push(&chunk)?,
Read::Spill(file) => car = car.spill(file)?,
}
}
car.finish()
});
let mut received = 0usize;
let read = async {
loop {
// the reader stopping on an error ends the download now, not when the next chunk
// shows up, which a stalled pds may never send
let chunk = tokio::select! {
chunk = resp.chunk() => chunk
.map_err(|e| BackfillError::Transport(e.to_string().into()))?,
() = tx.closed() => break,
};
let Some(chunk) = chunk else { break };
received += chunk.len();
if received > max_bytes {
return Ok(false);
}
if let Some(fetch) = &mut whole {
fetch.received(received, &tx).await?;
}
// a closed channel means the reader stopped on an error, which joining it reports
if tx.send(Read::Chunk(chunk)).await.is_err() {
break;
}
}
Ok::<_, BackfillError>(true)
}
.await;
drop(tx);
let mut outcome = reader.await.into_diagnostic()?;
if let Ok(FullRepoOutcome::Car { budget: held, .. }) = &mut outcome {
*held = whole.map(|fetch| fetch.reservation);
}
match read? {
false => Ok(None),
true => Ok(Some(outcome?)),
}
}
/// the records a filtered car's scan reaches. a record whose $type isn't its collection was
/// dropped while the car streamed, so any leaf the scan finds missing is fetched again: by cid
/// if the pds has getBlocks, else from one whole car. leaves still missing after that weren't
/// in the car at all, and are skipped like a whole car's missing leaves are
#[allow(clippy::too_many_arguments)]
async fn filtered_records(
app_state: &AppState,
http: &ThrottledHttpClient,
did: &Did,
pds: &url::Url,
throttle: &ThrottleHandle,
tier: &RateTier,
repo: FilteredRepo,
mst_root: IpldCid,
) -> Result<FilteredRecords, BackfillError> {
let FilteredRepo {
root,
blocks,
mut scanner,
..
} = repo;
let mut can_get_blocks = true;
let mut scan_rounds = 0usize;
let leaves = loop {
let (next_scanner, scan) = tokio::task::spawn_blocking(move || {
let scan = scanner.scan(mst_root)?;
Ok::<_, miette::Report>((scanner, scan))
})
.await
.into_diagnostic()??;
scanner = next_scanner;
let missing = match scan {
Ok(scan) => break scan.leaves,
Err(missing) => missing,
};
if let Some(cid) = missing.iter().find(|cid| !scanner.was_evicted(cid)) {
return Err(miette::miette!("mst node {cid} missing from car").into());
}
scan_rounds += 1;
if scan_rounds > SPARSE_MAX_SCAN_ROUNDS {
return Err(
miette::miette!("filtered mst scan exceeded fetch round limit").into(),
);
}
debug!(
missing = missing.len(),
round = scan_rounds,
"fetching mst nodes the filtered car evicted"
);
if can_get_blocks {
match fetch_blocks(
http,
pds,
did,
&missing,
throttle,
tier,
app_state.verify_cids,
app_state.max_car_body_bytes,
)
.await
{
Ok(fetched_nodes) => {
if missing.iter().any(|cid| fetched_nodes.get(cid).is_some()) {
scanner.insert_blocks(fetched_nodes);
continue;
}
can_get_blocks = false;
}
Err(
err @ (BackfillError::Ratelimited
| BackfillError::PreemptivelyThrottled
| BackfillError::Transport(_)),
) => return Err(err),
Err(err) => {
debug!(%err, "getBlocks failed, streaming the whole car instead");
can_get_blocks = false;
}
}
}
let resumed = FilteredCar::resume(root, scanner, app_state.verify_cids);
scanner = refetch_filtered_car(app_state, http, did, pds, throttle, tier, resumed)
.await?
.scanner;
if let Some(cid) = missing.iter().find(|cid| scanner.is_pending(cid)) {
return Err(miette::miette!("mst node {cid} missing from car").into());
}
};
let mut missing = leaves
.iter()
.map(|(_, cid)| *cid)
.filter(|cid| blocks.get(cid).is_none())
.collect::<Vec<_>>();
missing.sort_unstable();
missing.dedup();
let mut fetched = Blocks::default();
let mut whole = None;
if !missing.is_empty() {
debug!(
missing = missing.len(),
"fetching records the filtered car dropped"
);
if can_get_blocks {
match fetch_blocks(
http,
pds,
did,
&missing,
throttle,
tier,
app_state.verify_cids,
app_state.max_car_body_bytes,
)
.await
{
Ok(blocks) => fetched = blocks,
// a pds that's slowing us down or can't be reached gets the task's backoff, not a
// whole getRepo on top
Err(
err @ (BackfillError::Ratelimited
| BackfillError::PreemptivelyThrottled
| BackfillError::Transport(_)),
) => return Err(err),
Err(err) => debug!(%err, "getBlocks failed, fetching the whole car instead"),
}
}
let still_missing = missing
.into_iter()
.filter(|cid| fetched.get(cid).is_none())
.collect::<HashSet<_>>();
if !still_missing.is_empty() {
let only = FilteredCar::only(root, still_missing, app_state.verify_cids);
whole = Some(
refetch_filtered_car(app_state, http, did, pds, throttle, tier, only)
.await?
.blocks,
);
}
}
Ok(FilteredRecords {
leaves,
blocks,
fetched,
whole,
})
}
/// streams a second car of the repo whose filtered car came from commit `root` into `car`. a
/// repo that moved on in between is retried, since its records may no longer match the scan
#[allow(clippy::too_many_arguments)]
async fn refetch_filtered_car(
app_state: &AppState,
http: &ThrottledHttpClient,
did: &Did,
pds: &url::Url,
throttle: &ThrottleHandle,
tier: &RateTier,
car: FilteredCar,
) -> Result<FilteredRepo, BackfillError> {
let max = app_state.max_car_body_bytes;
match fetch_full_repo_car(http, pds, did, throttle, tier, max, RepoCar::Filtered(car)).await? {
FullRepoOutcome::Filtered(repo) => Ok(*repo),
FullRepoOutcome::NotFound => Err(BackfillError::RepoNotFound),
// the retry's first getRepo sees the same status and records it
FullRepoOutcome::Inactive(status) => {
Err(miette::miette!("repo went {status:?} while its records were refetched").into())
}
FullRepoOutcome::Car { .. } => {
Err(miette::miette!("a filtered getRepo handed back the whole body").into())
}
}
}
/// 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 {
/// every block of the car, within the configured size ceiling, and the share of the
/// backfill budget it took
Car {
car: WholeRepo,
budget: Option<Reservation>,
},
/// the parts of the car a filtered backfill can use, kept as the body streamed past.
Filtered(Box<FilteredRepo>),
/// 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,
car: RepoCar,
) -> 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 stream_car(resp, max_body_bytes, car).await?.ok_or_else(|| {
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 Some(code) = xrpc_error_code(&body) {
match code.as_str() {
"RepoNotFound" => return Ok(FullRepoOutcome::NotFound),
"RepoDeactivated" => {
return Ok(FullRepoOutcome::Inactive(RepoStatus::Deactivated));
}
"RepoTakendown" => {
return Ok(FullRepoOutcome::Inactive(RepoStatus::Takendown));
}
"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 crate::backfill::test_util::{public_client, serve};
use crate::filter::FilterMode;
#[cfg(feature = "indexer_stream")]
use crate::ops;
use axum::Router;
use axum::routing::get;
use smol_str::SmolStr;
use std::net::SocketAddr;
/// spawns a server answering `com.atproto.sync.getRepo` with a fixed status and body.
async fn spawn_get_repo(status: StatusCode, body: Vec<u8>) -> SocketAddr {
let app = Router::new().route(
"/xrpc/com.atproto.sync.getRepo",
get(move || {
let body = body.clone();
async move { (status, body) }
}),
);
serve(app).await
}
/// spawns a server answering getRepo with `chunks` as a chunked body, then dropping the
/// connection partway if `cut` is set
async fn spawn_streamed_get_repo(chunks: Vec<Bytes>, cut: bool) -> SocketAddr {
let app = Router::new().route(
"/xrpc/com.atproto.sync.getRepo",
get(move || {
use futures::StreamExt;
let parts = futures::stream::iter(chunks.clone().into_iter().map(Ok));
// late enough that the status and the first chunks are already out, so the
// client is reading the body when it breaks rather than still waiting on headers
let cut = futures::stream::iter(cut.then_some(())).then(|()| async {
tokio::time::sleep(std::time::Duration::from_millis(100)).await;
Err(std::io::Error::other("connection dropped"))
});
async move { axum::body::Body::from_stream(parts.chain(cut)) }
}),
);
serve(app).await
}
/// a car of just its commit, or of a commit whose bytes don't match its cid
async fn commit_only_car(forged: bool) -> Bytes {
let commit = b"commit".to_vec();
let cid = crate::car::compute_cid(&commit);
let mut bytes = Vec::new();
let mut writer =
crate::car::CarWriter::new(crate::car::CarHeader::new_v1(vec![cid]), &mut bytes);
let body: &[u8] = if forged { b"forged" } else { &commit };
writer.write(cid, body).await.unwrap();
writer.finish().await.unwrap();
Bytes::from(bytes)
}
fn whole_car(budget: &BackfillBudget) -> RepoCar {
RepoCar::whole(true, budget.clone(), Spills::never())
}
fn tangled_only() -> FilteredCar {
let mut filter = crate::filter::FilterConfig::new(FilterMode::Filter);
filter.collections = vec![SmolStr::new("sh.tangled.repo")];
FilteredCar::new(Arc::new(filter), true).unwrap()
}
async fn fetch(
pds: &SocketAddr,
max_body_bytes: usize,
) -> Result<FullRepoOutcome, BackfillError> {
fetch_with(
pds,
max_body_bytes,
whole_car(&BackfillBudget::new(usize::MAX)),
)
.await
}
async fn fetch_with(
pds: &SocketAddr,
max_body_bytes: usize,
car: RepoCar,
) -> Result<FullRepoOutcome, BackfillError> {
let throttler = crate::util::throttle::Throttler::new(10);
let (public_pds, http) = public_client(*pds, &throttler);
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,
car,
)
.await
}
#[tokio::test]
async fn streamed_get_repo_keeps_a_whole_car() {
let car = commit_only_car(false).await;
let pds =
spawn_streamed_get_repo(car.chunks(7).map(Bytes::copy_from_slice).collect(), false)
.await;
match fetch_with(&pds, 1024, RepoCar::Filtered(tangled_only()))
.await
.unwrap()
{
FullRepoOutcome::Filtered(repo) => assert_eq!(repo.stats.kept_blocks, 1),
other => panic!("expected Filtered, got {other:?}"),
}
}
#[tokio::test]
async fn streamed_get_repo_maps_a_dropped_connection_to_transport() {
let car = commit_only_car(false).await;
let half = car.slice(..car.len() / 2);
let pds = spawn_streamed_get_repo(vec![half], true).await;
let err = fetch_with(&pds, 1024, RepoCar::Filtered(tangled_only()))
.await
.unwrap_err();
assert!(matches!(err, BackfillError::Transport(_)), "{err}");
}
#[tokio::test]
async fn streamed_get_repo_rejects_a_forged_block_without_blaming_the_network() {
let pds = spawn_streamed_get_repo(vec![commit_only_car(true).await], false).await;
let err = fetch_with(&pds, 1024, RepoCar::Filtered(tangled_only()))
.await
.unwrap_err();
assert!(matches!(err, BackfillError::Generic(_)), "{err}");
}
#[tokio::test]
async fn streamed_get_repo_stops_at_the_body_limit() {
let car = commit_only_car(false).await;
let max = car.len() - 1;
// a body that says up front how big it is, and one that only shows it as it arrives
let sized = spawn_get_repo(StatusCode::OK, car.to_vec()).await;
let chunked =
spawn_streamed_get_repo(car.chunks(7).map(Bytes::copy_from_slice).collect(), false)
.await;
for pds in [sized, chunked] {
let err = fetch_with(&pds, max, RepoCar::Filtered(tangled_only()))
.await
.unwrap_err();
assert!(
err.to_string().contains("exceeded max body size"),
"{pds}: {err}"
);
}
}
#[cfg(feature = "indexer_stream")]
#[tokio::test]
async 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();
let staged = did.clone();
ops::announce(&app_state.db, move |events| {
BackfillImported { before, after }.stage(&staged, events)
})
.await;
let crate::types::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 car = commit_only_car(false).await;
let pds = spawn_get_repo(StatusCode::OK, car.to_vec()).await;
let whole = crate::car::parse_car(car, true).unwrap();
match fetch(&pds, 1024).await.unwrap() {
FullRepoOutcome::Car { car, .. } => {
assert_eq!(car.root, whole.root);
let WholeBlocks::Memory(blocks) = car.blocks else {
panic!("a car under the threshold spilled");
};
assert_eq!(blocks.map(), whole.blocks.map());
}
other => panic!("expected Car, got {other:?}"),
}
}
#[tokio::test]
async fn full_get_repo_checks_blocks_as_they_arrive() {
// a forged block, then a body that never ends: only a reader that checks each block
// as it comes in gets to answer
let car = commit_only_car(true).await;
let app = Router::new().route(
"/xrpc/com.atproto.sync.getRepo",
get(move || {
use futures::StreamExt;
let first = futures::stream::iter([Ok::<_, std::io::Error>(car.clone())]);
async move { axum::body::Body::from_stream(first.chain(futures::stream::pending())) }
}),
);
let pds = serve(app).await;
for car in [
whole_car(&BackfillBudget::new(usize::MAX)),
RepoCar::Filtered(tangled_only()),
] {
let fetch = fetch_with(&pds, 1024, car);
let fetched = tokio::time::timeout(std::time::Duration::from_secs(5), fetch)
.await
.expect("the forged block ends the fetch before the body does");
let err = fetched.unwrap_err();
assert!(err.to_string().contains("CAR block CID mismatch"), "{err}");
}
}
#[tokio::test]
async fn a_whole_car_holds_its_budget_until_dropped() {
let car = commit_only_car(false).await;
let budget = BackfillBudget::new(usize::MAX);
// one sized body, and one only seen as it arrives
let sized = spawn_get_repo(StatusCode::OK, car.to_vec()).await;
let chunked =
spawn_streamed_get_repo(car.chunks(7).map(Bytes::copy_from_slice).collect(), false)
.await;
for pds in [sized, chunked] {
let outcome = fetch_with(&pds, 1024, whole_car(&budget)).await.unwrap();
assert!(matches!(
outcome,
FullRepoOutcome::Car {
budget: Some(_),
..
}
));
assert_eq!(budget.used(), peak_for(car.len()), "{pds}");
drop(outcome);
assert_eq!(budget.used(), 0, "{pds}");
}
}
#[tokio::test]
async fn a_spilled_car_holds_its_index_and_a_step_not_the_car() -> Result<()> {
let car = commit_only_car(false).await;
let budget = BackfillBudget::new(usize::MAX);
let tmp = tempfile::tempdir().into_diagnostic()?;
let spills = Spills::open(tmp.path().join("spill"), 0, 1)?;
let sized = spawn_get_repo(StatusCode::OK, car.to_vec()).await;
let chunked =
spawn_streamed_get_repo(car.chunks(7).map(Bytes::copy_from_slice).collect(), false)
.await;
for pds in [sized, chunked] {
let spilling = RepoCar::whole(false, budget.clone(), spills.clone());
let outcome = fetch_with(&pds, 1024, spilling).await.unwrap();
let FullRepoOutcome::Car { car: repo, .. } = &outcome else {
panic!("expected Car, got {outcome:?}");
};
assert!(matches!(repo.blocks, WholeBlocks::Spilled(_)), "{pds}");
assert!(repo.blocks.read(&repo.root)?.is_some());
assert_eq!(budget.used(), spilled_peak_for(car.len()), "{pds}");
drop(outcome);
assert_eq!(budget.used(), 0, "{pds}");
}
Ok(())
}
#[tokio::test]
async fn a_car_that_passes_the_threshold_partway_spills_what_it_read_so_far() -> Result<()> {
let blocks = (0..20)
.map(|i| {
let body = serde_ipld_dagcbor::to_vec(&serde_json::json!({ "n": i })).unwrap();
(crate::car::compute_cid(&body), body)
})
.collect::<Vec<_>>();
let mut car = Vec::new();
let mut writer =
crate::car::CarWriter::new(crate::car::CarHeader::new_v1(vec![blocks[0].0]), &mut car);
for (cid, body) in &blocks {
writer.write(*cid, body).await.into_diagnostic()?;
}
writer.finish().await.into_diagnostic()?;
// no length up front, so the car only turns out too big halfway through
let chunked =
spawn_streamed_get_repo(car.chunks(7).map(Bytes::copy_from_slice).collect(), false)
.await;
let budget = BackfillBudget::new(usize::MAX);
let tmp = tempfile::tempdir().into_diagnostic()?;
let spills = Spills::open(tmp.path().join("spill"), car.len() / 2, 1)?;
let spilling = RepoCar::whole(false, budget.clone(), spills);
let outcome = fetch_with(&chunked, 4096, spilling).await.unwrap();
let FullRepoOutcome::Car { car: repo, .. } = &outcome else {
panic!("expected Car, got {outcome:?}");
};
assert!(matches!(repo.blocks, WholeBlocks::Spilled(_)));
assert_eq!(repo.blocks.len(), blocks.len());
for (cid, body) in &blocks {
assert_eq!(repo.blocks.read(cid)?.as_deref(), Some(body.as_slice()));
}
assert_eq!(budget.used(), spilled_peak_for(car.len()));
Ok(())
}
#[tokio::test]
async fn a_whole_car_that_fails_partway_gives_its_budget_back() {
let budget = BackfillBudget::new(usize::MAX);
let car = commit_only_car(false).await;
let half = car.slice(..car.len() / 2);
let dropped = spawn_streamed_get_repo(vec![half], true).await;
let forged = spawn_streamed_get_repo(vec![commit_only_car(true).await], false).await;
for pds in [dropped, forged] {
fetch_with(&pds, 1024, whole_car(&budget))
.await
.unwrap_err();
assert_eq!(budget.used(), 0, "{pds}");
}
}
#[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:?}"),
}
}
#[tokio::test]
async fn filtered_records_recovers_evicted_mst_nodes_or_fails_when_un_evicted() -> Result<()> {
use crate::car::compute_cid;
use std::sync::atomic::{AtomicUsize, Ordering};
let did = Did::new_static("did:plc:aaaaaaaaaaaaaaaaaaaaaaaa")?;
let record_body = serde_ipld_dagcbor::to_vec(&serde_json::json!({
"$type": "sh.tangled.repo",
"name": "core",
}))
.into_diagnostic()?;
let record_cid = compute_cid(&record_body);
let child_node = crate::mst::tests::node(
vec![("sh.tangled.repo/core", record_cid, None)],
None,
);
let child_bytes = serde_ipld_dagcbor::to_vec(&child_node).into_diagnostic()?;
let child_cid = compute_cid(&child_bytes);
let root_node = crate::mst::tests::node(
vec![("zz.other/1", record_cid, None)],
Some(child_cid),
);
let root_bytes = serde_ipld_dagcbor::to_vec(&root_node).into_diagnostic()?;
let mst_root = compute_cid(&root_bytes);
let commit = crate::mst::AtpCommit {
did: did.clone(),
version: 3,
data: mst_root,
rev: "3l2abcdefgh22".parse()?,
prev: None,
sig: Bytes::new(),
};
let commit_bytes = commit.to_cbor()?;
let commit_cid = commit.to_cid()?;
// `child_cid` arrives before `mst_root`, so a 0-byte parked budget evicts it
let mut car_bytes = Vec::new();
let mut writer = crate::car::CarWriter::new(
crate::car::CarHeader::new_v1(vec![commit_cid]),
&mut car_bytes,
);
writer.write(commit_cid, &commit_bytes).await.into_diagnostic()?;
writer.write(child_cid, &child_bytes).await.into_diagnostic()?;
writer.write(record_cid, &record_body).await.into_diagnostic()?;
writer.write(mst_root, &root_bytes).await.into_diagnostic()?;
writer.finish().await.into_diagnostic()?;
let car_bytes = Bytes::from(car_bytes);
let mut blocks_reply = Vec::new();
let mut blocks_writer = crate::car::CarWriter::new(
crate::car::CarHeader::new_v1(vec![child_cid]),
&mut blocks_reply,
);
blocks_writer
.write(child_cid, &child_bytes)
.await
.into_diagnostic()?;
blocks_writer.finish().await.into_diagnostic()?;
let blocks_reply = Bytes::from(blocks_reply);
let mut filter = crate::filter::FilterConfig::new(FilterMode::Filter);
filter.collections = vec![SmolStr::new("sh.tangled.repo")];
let filter = Arc::new(filter);
let tmp = tempfile::tempdir().into_diagnostic()?;
let app_state = AppState::new(&crate::config::Config {
database_path: tmp.path().to_path_buf(),
..Default::default()
})?;
for serve_get_blocks in [true, false] {
let repo_hits = Arc::new(AtomicUsize::new(0));
let block_hits = Arc::new(AtomicUsize::new(0));
let car_body = car_bytes.clone();
let blk_body = blocks_reply.clone();
let repo_hits_route = repo_hits.clone();
let block_hits_route = block_hits.clone();
let app = Router::new()
.route(
"/xrpc/com.atproto.sync.getRepo",
get(move || {
repo_hits_route.fetch_add(1, Ordering::SeqCst);
let body = car_body.clone();
async move { (StatusCode::OK, body) }
}),
)
.route(
"/xrpc/com.atproto.sync.getBlocks",
get(move || {
block_hits_route.fetch_add(1, Ordering::SeqCst);
let body = blk_body.clone();
async move {
if serve_get_blocks {
(StatusCode::OK, body)
} else {
(StatusCode::NOT_FOUND, Bytes::new())
}
}
}),
);
let addr = serve(app).await;
let throttler = crate::util::throttle::Throttler::new(10);
let (public_pds, http) = public_client(addr, &throttler);
let throttle = throttler.get_handle(&public_pds).await;
let mut filtered =
FilteredCar::with_parked_budget(filter.clone(), 0, true).unwrap();
filtered.push(&car_bytes)?;
let repo = filtered.finish()?;
assert!(repo.scanner.evicted_any());
let out = filtered_records(
&app_state,
&http,
&did,
&public_pds,
&throttle,
&RateTier::trusted(),
repo,
mst_root,
)
.await?;
assert_eq!(
out.leaves,
vec![(SmolStr::new("sh.tangled.repo/core"), record_cid)]
);
assert_eq!(block_hits.load(Ordering::SeqCst), 1);
assert_eq!(
repo_hits.load(Ordering::SeqCst),
usize::from(!serve_get_blocks)
);
}
// a missing mst node (even when another out-of-order node was evicted) fails immediately
// without calling getBlocks or refetching the car
let stale_bytes = serde_ipld_dagcbor::to_vec(&crate::mst::tests::node(
vec![("zz.other/1", record_cid, None)],
Some(record_cid),
))
.into_diagnostic()?;
let stale_cid = crate::car::compute_cid(&stale_bytes);
let mut missing_car = Vec::new();
let mut missing_writer = crate::car::CarWriter::new(
crate::car::CarHeader::new_v1(vec![commit_cid]),
&mut missing_car,
);
missing_writer
.write(stale_cid, &stale_bytes)
.await
.into_diagnostic()?;
missing_writer
.write(commit_cid, &commit_bytes)
.await
.into_diagnostic()?;
missing_writer
.write(mst_root, &root_bytes)
.await
.into_diagnostic()?;
missing_writer.finish().await.into_diagnostic()?;
let mut filtered = FilteredCar::with_parked_budget(filter, 0, true).unwrap();
filtered.push(&missing_car)?;
let repo = filtered.finish()?;
assert!(repo.scanner.evicted_any());
assert!(!repo.scanner.was_evicted(&child_cid));
let addr = spawn_get_repo(StatusCode::OK, car_bytes.to_vec()).await;
let throttler = crate::util::throttle::Throttler::new(10);
let (public_pds, http) = public_client(addr, &throttler);
let throttle = throttler.get_handle(&public_pds).await;
let err = filtered_records(
&app_state,
&http,
&did,
&public_pds,
&throttle,
&RateTier::trusted(),
repo,
mst_root,
)
.await
.err()
.unwrap();
assert!(err.to_string().contains("missing from car"), "{err}");
Ok(())
}
}