[backfill] batch full CAR block access
4 unmodified lines56789101112131415161718192021222324252627282930313233343536371138394014414243444546472048495051525354555682 unmodified lines13914014114214314414514614714814915015115215315415515615715815916016116216316416516616716811 unmodified lines4 unmodified linesuse cid::Cid as IpldCid;use miette::{IntoDiagnostic, Result};
#[cfg_attr(not(feature = "indexer"), allow(dead_code))]#[derive(Debug)]pub(crate) struct ParsedCar { pub root: IpldCid, pub blocks: BTreeMap<IpldCid, Bytes>,}
#[cfg_attr(not(feature = "indexer"), allow(dead_code))]pub(crate) fn parse_car(data: Bytes) -> Result<ParsedCar> { let (header_start, blocks_start) = car_header_bounds(&data)?; let header = iroh_car::CarHeader::decode(&data[header_start..blocks_start]).into_diagnostic()?; let root = header .roots() .first() .copied() .ok_or_else(|| miette::miette!("CAR file has no roots"))?; let blocks = parse_car_blocks_from(data, blocks_start)?;
Ok(ParsedCar { root, blocks })}
#[cfg_attr(not(feature = "indexer"), allow(dead_code))]pub(crate) fn parse_car_blocks(data: Bytes) -> Result<BTreeMap<IpldCid, Bytes>> { let (_, blocks_start) = car_header_bounds(&data)?; parse_car_blocks_from(data, blocks_start)}
fn car_header_bounds(data: &[u8]) -> Result<(usize, usize)> { let mut offset = 0; let Some(header_len) = read_uvarint(&data, &mut offset)? else { let Some(header_len) = read_uvarint(data, &mut offset)? else { return Err(miette::miette!("empty CAR file")); }; let header_end = offset let header_start = offset; let header_end = header_start .checked_add(header_len) .ok_or_else(|| miette::miette!("CAR header length overflow"))?; if header_end > data.len() { return Err(miette::miette!("truncated CAR header")); } offset = header_end;
Ok((header_start, header_end))}
fn parse_car_blocks_from(data: Bytes, mut offset: usize) -> Result<BTreeMap<IpldCid, Bytes>> { let mut blocks = BTreeMap::new(); while let Some(section_len) = read_uvarint(&data, &mut offset)? { let section_end = offset .checked_add(section_len)82 unmodified lines let blocks = parse_car_blocks(buf.into()).unwrap(); assert_eq!(blocks.get(&cid).unwrap().as_ref(), b"block"); }
#[tokio::test] async fn parses_rooted_car_with_zero_copy_payloads() { let root = cid(1); let block = cid(2); let mut buf = Vec::new(); let header = iroh_car::CarHeader::new_v1(vec![root]); let mut writer = iroh_car::CarWriter::new(header, &mut buf); writer.write(block, b"record".to_vec()).await.unwrap(); writer.finish().await.unwrap();
let data = Bytes::from(buf); let data_range = data.as_ptr() as usize..data.as_ptr() as usize + data.len(); let parsed = parse_car(data).unwrap(); let payload = parsed.blocks.get(&block).unwrap(); let payload_range = payload.as_ptr() as usize..payload.as_ptr() as usize + payload.len();
assert_eq!(parsed.root, root); assert_eq!(payload.as_ref(), b"record"); assert!(data_range.start <= payload_range.start); assert!(payload_range.end <= data_range.end); }
#[test] fn validate_block_cids_rejects_mismatched_cid() { let blocks = BTreeMap::from([(cid(1), Bytes::from_static(b"forged"))]); let err = validate_block_cids(&blocks).unwrap_err();11 unmodified lines150 unmodified linesExpand all15115215315415515615715815916016116216316416516616716816917017117215515615715815916016116216316416516616716816917017117217317417517682 unmodified lines25926026126226326426526626726826927027127227327426326426526626726827627727827928028126927027127227327427527627727827928028128228328428328428528628728528628728828929029129229323 unmodified lines31731831932031932032132132232332432532632732832933033133233333033133233323 unmodified lines3573583593603653613623633643653663673683693703713728 unmodified lines38138238338438538638738838938538638738838939025 unmodified lines41641741841942342442542642742042142242342435 unmodified lines46046146246347147247346446546646751 unmodified lines51952052152253353453553653753853954054154254354454554654754854952352452552652752852953053153253353453553653753853954054154254354413 unmodified lines150 unmodified linesExpand all let metadata_key = keys::repo_metadata_key(did); let app_state_clone = app_state.clone(); let did = did.clone(); let pending_key = pending_key.clone(); app_state_clone.db.run(move |db| { let mut txn = DbTxn::new(db); let applied = txn.transition_pending_key( &did, pending_key.as_ref(), GaugeState::Synced, )?; txn.batch .remove(&db.indexer.pending, pending_key); if applied { txn.batch.remove(&db.repos, &did_key); txn.batch .remove(&db.repo_metadata, &metadata_key); txn.counts.add_repos(-1); } txn.commit() }) .await?; app_state_clone .db .run(move |db| { let mut txn = DbTxn::new(db); let applied = txn.transition_pending_key( &did, pending_key.as_ref(), GaugeState::Synced, )?; txn.batch.remove(&db.indexer.pending, pending_key); if applied { txn.batch.remove(&db.repos, &did_key); txn.batch.remove(&db.repo_metadata, &metadata_key); txn.counts.add_repos(-1); } txn.commit() }) .await?;
return Ok(None); } Ok(SparseBackfillResult::Skipped) => {82 unmodified lines
let app_state_clone = app_state.clone(); let did = did.clone(); let pending_key = pending_key.clone(); app_state_clone.db.run(move |db| { let mut txn = DbTxn::new(db); let applied = txn.transition_pending_key( &did, pending_key.as_ref(), GaugeState::Resync(None), )?; txn.batch.remove(&db.indexer.pending, pending_key.clone()); if applied { crate::db::Db::update_repo_state( &mut txn.batch, &db.repos, app_state_clone .db .run(move |db| { let mut txn = DbTxn::new(db); let applied = txn.transition_pending_key( &did, move |state, (key, batch)| { state.active = false; state.status = status; batch.insert(&db.indexer.resync, key, resync_bytes); Ok((true, ())) }, pending_key.as_ref(), GaugeState::Resync(None), )?; txn.batch.remove(&db.indexer.pending, pending_key.clone()); if applied { crate::db::Db::update_repo_state( &mut txn.batch, &db.repos, &did, move |state, (key, batch)| { state.active = false; state.status = status; batch.insert(&db.indexer.resync, key, resync_bytes); Ok((true, ())) }, )?; } txn.commit()?; Ok::<_, miette::Report>(()) }) .await?; } txn.commit()?; Ok::<_, miette::Report>(()) }) .await?;
return Ok(None); }
23 unmodified lines // (finding d87fbd22402c81919fceb9cdfee53cd4)
// 3. import repo let start = Instant::now(); let parsed = jacquard_repo::car::reader::parse_car_bytes(&car_bytes.body) .await .into_diagnostic()?; let parsed = crate::car::parse_car(car_bytes.body.into())?; trace!(elapsed = %start.elapsed().as_secs_f32(), "parsed car");
let start = Instant::now(); let root_cid = parsed.root; if app_state.verify_cids { crate::car::validate_block_cids(&parsed.blocks)?; } let store = Arc::new(MemoryBlockStore::new_from_blocks(parsed.blocks)); // parsed.blocks was moved into the store; parsed.root is now root_cid. drop the raw // CAR bytes here so we don't hold two copies of the block data (raw + parsed) for // the entire duration of the spawn_blocking call below. drop(car_bytes); trace!( blocks = store.len(), elapsed = ?start.elapsed(), "stored blocks in memory"23 unmodified lines }
let root_commit = Commit::from(root_commit);
// 5. walk mst // 5. walk mst and fetch every record block under one store lock let start = Instant::now(); let mst: Mst<MemoryBlockStore> = Mst::load(store, root_commit.data, None); let leaves = mst.leaves().await.into_diagnostic()?; let leaf_cids = leaves.iter().map(|(_, cid)| *cid).collect::<Vec<_>>(); let leaf_blocks = mst.storage().get_many(&leaf_cids).await.into_diagnostic()?; let records = leaves.into_iter().zip(leaf_blocks); drop(mst); trace!(elapsed = %start.elapsed().as_secs_f32(), "walked mst");
// 6. insert records into db let start = Instant::now();8 unmodified lines let mut count = 0; let mut collection_counts: HashMap<SmolStr, u64> = HashMap::new(); let mut txn = DbTxn::new(&app_state.db); let mut record_txn = txn.backfill_records(&app_state, &rev, &did); // clone the Arc so we hold an independent reference to the block store, // allowing mst (and its entire loaded node tree) to be freed immediately // rather than surviving until the end of spawn_blocking. let store = mst.storage().clone(); drop(mst); // the MST and non-record blocks were released before entering this blocking // persistence phase; records contains only leaf metadata and payload slices.
let prefix = keys::record_prefix_did(&did); let mut existing_cids: HashMap<(SmolStr, DbRkey), SmolStr> = HashMap::new();
25 unmodified lines }
let mut signal_seen = filter.mode == FilterMode::Full || filter.signals.is_empty();
for (key, cid) in leaves { let val_bytes = tokio::runtime::Handle::current() .block_on(store.get(&cid)) .into_diagnostic()?;
for ((key, cid), val_bytes) in records { if let Some(val) = val_bytes { let (collection, rkey) = ops::parse_path(&key)?;
if !filter.matches_collection(collection) {35 unmodified lines count += 1; } }
// all blocks have been read; free the MemoryBlockStore now so it does not // survive through the final batch commit or any fjall backpressure stall. drop(store);
// remove any remaining existing records (they weren't in the new MST) for ((collection, rkey), cid) in existing_cids { trace!(collection = %collection, rkey = %rkey, cid = %cid, "remove existing record");51 unmodified lines let metadata_key = keys::repo_metadata_key(did); let backfill_pending_key = pending_key.clone(); let app_state = app_state.clone(); let did = did.clone(); app_state.db.run(move |db| { let mut txn = DbTxn::new(db); let applied = txn.transition_pending_key( &did, backfill_pending_key.as_ref(), GaugeState::Synced, )?; txn.batch .remove(&db.indexer.pending, backfill_pending_key); if applied { txn.batch.remove(&db.repos, &did_key); txn.batch.remove(&db.repo_metadata, &metadata_key); txn.counts.add_repos(-1); } txn.commit() }) .await?; app_state .db .run(move |db| { let mut txn = DbTxn::new(db); let applied = txn.transition_pending_key( &did, backfill_pending_key.as_ref(), GaugeState::Synced, )?; txn.batch.remove(&db.indexer.pending, backfill_pending_key); if applied { txn.batch.remove(&db.repos, &did_key); txn.batch.remove(&db.repo_metadata, &metadata_key); txn.counts.add_repos(-1); } txn.commit() }) .await?; return Ok(None); };
trace!(ops = count, elapsed = %start.elapsed().as_secs_f32(), "did ops");13 unmodified lines