From 22b2e32450ebebfd26f4685b3f52158b3d6123a0 Mon Sep 17 00:00:00 2001 From: dawn <90008@klbr.net> Date: Mon, 5 Oct 2026 21:11:34 +0300 Subject: [PATCH] [backfill] stop a whole-car refetch at the first block of another commit the refetch for records a filtered car dropped downloaded the whole second car before noticing the repo had moved on, which for a hold-sized repo is another gigabyte thrown away. FilteredCar::only now takes the commit it expects and fails on the first block of any other, since the header comes first. a repo gone from the pds by the refetch is retried as RepoNotFound like the first getRepo, and one that went inactive says so instead of "changed state". --- src/backfill/filtered_car.rs | 59 +++++++++++++++++++++++++++------- src/backfill/worker/process.rs | 20 ++++++------ src/backfill/worker/task.rs | 24 ++++++++++++++ 3 files changed, 82 insertions(+), 21 deletions(-) diff --git a/src/backfill/filtered_car.rs b/src/backfill/filtered_car.rs index 79c3a0d..42f70ca 100644 --- a/src/backfill/filtered_car.rs +++ b/src/backfill/filtered_car.rs @@ -34,8 +34,12 @@ enum Wanted { ranges: Vec, ruled_out: RuledOut, }, - /// these and nothing else, for going back for what an earlier car left behind - Only(HashSet), + /// these and nothing else from a car of commit `root`, for going back for what an earlier + /// car of it left behind + Only { + root: IpldCid, + cids: HashSet, + }, } #[derive(Debug, Default, Clone, Copy)] @@ -76,8 +80,8 @@ impl FilteredCar { /// keeps the commit and `cids`, so a second car costs what's missing rather than the /// whole repo - pub(crate) fn only(cids: HashSet, verify_cids: bool) -> Self { - Self::keeping(Wanted::Only(cids), verify_cids) + pub(crate) fn only(root: IpldCid, cids: HashSet, verify_cids: bool) -> Self { + Self::keeping(Wanted::Only { root, cids }, verify_cids) } fn keeping(wanted: Wanted, verify_cids: bool) -> Self { @@ -93,10 +97,8 @@ impl FilteredCar { pub(crate) fn push(&mut self, chunk: &[u8]) -> Result<()> { let keep = &mut self.keep; - self.stream.push(chunk, |root, cid, bytes| { - keep.offer(root, cid, bytes); - Ok(()) - }) + self.stream + .push(chunk, |root, cid, bytes| keep.offer(root, cid, bytes)) } pub(crate) fn finish(self) -> Result { @@ -110,7 +112,10 @@ impl FilteredCar { Wanted::Filter { ranges, ruled_out, .. } => (ranges, ruled_out), - Wanted::Only(_) => (Vec::new(), RuledOut::default()), + Wanted::Only { root, .. } => { + moved_on(&root, &car.root)?; + (Vec::new(), RuledOut::default()) + } }; Ok(FilteredRepo { root: car.root, @@ -123,7 +128,12 @@ impl FilteredCar { } impl Keep { - fn offer(&mut self, root: &IpldCid, cid: IpldCid, bytes: &[u8]) { + fn offer(&mut self, root: &IpldCid, cid: IpldCid, bytes: &[u8]) -> Result<()> { + // the header comes first, so a car of another commit stops at its first block + // rather than after downloading the whole repo + if let Wanted::Only { root: expected, .. } = &self.wanted { + moved_on(expected, root)?; + } self.stats.blocks += 1; self.stats.bytes += bytes.len(); let wanted = cid == *root || self.wanted.keeps(&cid, bytes); @@ -134,13 +144,21 @@ impl Keep { // copied, so a kept block doesn't pin the chunk it arrived in self.blocks.insert(cid, Bytes::copy_from_slice(bytes)); } + Ok(()) + } +} + +fn moved_on(expected: &IpldCid, root: &IpldCid) -> Result<()> { + if root != expected { + miette::bail!("repo moved on: this car is of commit {root}, not {expected}"); } + Ok(()) } impl Wanted { fn keeps(&mut self, cid: &IpldCid, bytes: &[u8]) -> bool { match self { - Wanted::Only(cids) => cids.contains(cid), + Wanted::Only { cids, .. } => cids.contains(cid), Wanted::Filter { filter, ranges, @@ -263,7 +281,8 @@ mod tests { let records = mixed_repo(); let (car, _) = repo_car(&records).await; let asked = [&records[3].1, &records[1999].1].map(|body| compute_cid(body).unwrap()); - let mut only = FilteredCar::only(asked.into_iter().collect(), true); + let root = crate::car::parse_car(car.clone(), false).unwrap().root; + let mut only = FilteredCar::only(root, asked.into_iter().collect(), true); for piece in car.chunks(64) { only.push(piece).unwrap(); } @@ -277,6 +296,22 @@ mod tests { assert_eq!(repo.ruled_out.len(), 0); } + #[tokio::test] + async fn only_stops_at_the_first_block_of_another_commit() { + let (car, _) = repo_car(&mixed_repo()).await; + let other = compute_cid(b"another commit").unwrap(); + let mut only = FilteredCar::only(other, HashSet::new(), true); + let pieces = car.chunks(64).collect::>(); + let failed_at = pieces + .iter() + .position(|piece| only.push(piece).is_err()) + .expect("a car of another commit is refused"); + assert!( + failed_at < pieces.len() / 100, + "stopped at piece {failed_at}" + ); + } + #[test] fn takes_every_block_when_the_filter_takes_every_collection() { assert!(FilteredCar::new(filter(&[]), true).is_none()); diff --git a/src/backfill/worker/process.rs b/src/backfill/worker/process.rs index cca6a61..8b5e7ad 100644 --- a/src/backfill/worker/process.rs +++ b/src/backfill/worker/process.rs @@ -751,16 +751,18 @@ async fn blocks_from_whole_car( cids: HashSet, ) -> Result, BackfillError> { let max = app_state.max_car_body_bytes; - let only = FilteredCar::only(cids, app_state.verify_cids); - let FullRepoOutcome::Filtered(repo) = - fetch_full_repo_car(http, pds, did, throttle, tier, max, Some(only)).await? - else { - return Err(miette::miette!("repo changed state while its records were refetched").into()); - }; - if repo.root != root { - return Err(miette::miette!("repo moved on while its records were refetched").into()); + let only = FilteredCar::only(root, cids, app_state.verify_cids); + match fetch_full_repo_car(http, pds, did, throttle, tier, max, Some(only)).await? { + FullRepoOutcome::Filtered(repo) => Ok(repo.blocks), + 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()) + } } - Ok(repo.blocks) } /// upper bound on a buffered xrpc error body. error responses are tiny json objects; this only diff --git a/src/backfill/worker/task.rs b/src/backfill/worker/task.rs index 4f44916..5263316 100644 --- a/src/backfill/worker/task.rs +++ b/src/backfill/worker/task.rs @@ -1126,6 +1126,30 @@ mod tests { Ok(()) } + #[tokio::test] + async fn a_repo_gone_before_the_refetch_is_retried_as_not_found() -> miette::Result<()> { + let (car, _) = car_with_a_misfiled_record().await?; + let fixture = Fixture::new().await?; + only_collection(&fixture, WANTED); + fixture.seed(&RepoState::backfilling().into_static(), 9, &[])?; + fixture.pds.serve_car(car.bytes).await; + + let result = fixture + .run_changing_at(&fixture.pds.get_blocks, keys::pending_key(9), async { + *fixture.pds.response.write().await = + (StatusCode::BAD_REQUEST, Bytes::from_static(NOT_FOUND_BODY)); + Ok(()) + }) + .await?; + + assert!( + matches!(result, Err(BackfillError::RepoNotFound)), + "{result:?}" + ); + assert!(fixture.record_rkeys()?.is_empty()); + Ok(()) + } + #[tokio::test] async fn record_filed_under_another_type_comes_back_from_a_whole_car() -> miette::Result<()> { // the stub has no getBlocks, like the atcr hold, so the dropped record can only come -- 2.54.0