Something went wrong. Try again.
Monorepo for Tangled
Something went wrong. Try again.
31 kB · 993 lines
Rust
at forward-ips
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819820821822823824825826827828829830831832833834835836837838839840841842843844845846847848849850851852853854855856857858859860861862863864865866867868869870871872873874875876877878879880881882883884885886887888889890891892893894895896897898899900901902903904905906907908909910911912913914915916917918919920921922923924925926927928929930931932933934935936937938939940941942943944945946947948949950951952953954955956957958959960961962963964965966967968969970971972973974975976977978979980981982983984985986987988989990991992993994use std::collections::{HashMap, HashSet};use std::error::Error;use std::io::Write;use std::path::{Path, PathBuf};use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering};use std::sync::{Arc, mpsc};use std::time::{Duration, Instant};
use gix::ObjectId;use gix::prelude::FindExt;use gix::progress::Discard;use gix_pack::data::{Version, output};use knot_types::{ObjectCount, Oid};
use crate::error::{PackError, PackLimit};use crate::ids::{Crc32, MaxObjectBytes, PackOffset};use crate::meter::{self, PackLimits, check_depth};use crate::resolve;
type OidStream = Box<dyn Iterator<Item = Result<ObjectId, Box<dyn Error + Send + Sync>>> + Send>;
fn odb_at(objects_dir: &Path, kind: gix::hash::Kind) -> Result<gix::odb::Handle, PackError> { gix::odb::at_opts( objects_dir, std::iter::empty(), gix::odb::store::init::Options { object_hash: kind, ..Default::default() }, ) .map_err(|error| PackError::Pack(error.to_string()))}
pub fn write_pack( objects_dir: &Path, oids: Vec<Oid>, thin_bases: Option<&HashSet<Oid>>, out: &mut dyn Write, kind: gix::hash::Kind,) -> Result<(), PackError> { let mut odb = odb_at(objects_dir, kind)?; odb.prevent_pack_unload(); odb.refresh_never();
let interrupt = AtomicBool::new(false); let permitted_bases: Option<HashSet<ObjectId>> = thin_bases.map(|bases| bases.iter().map(|oid| oid.object_id()).collect()); let oids: OidStream = Box::new(oids.into_iter().map(|oid| Ok(oid.object_id())));
let (counts, _) = output::count::objects( odb.clone(), oids, &Discard, &interrupt, output::count::objects::Options { thread_limit: knot_resource::gix_thread_limit().map(knot_resource::ThreadCount::get), chunk_size: 50, input_object_expansion: output::count::objects::ObjectExpansion::AsIs, }, ) .map_err(|error| PackError::Pack(error.to_string()))?;
write_counts(counts, odb, permitted_bases, out, kind)}
pub struct ExpandedPack { counts: Vec<output::Count>, odb: gix::odb::Handle, kind: gix::hash::Kind,}
impl ExpandedPack { pub fn len(&self) -> usize { self.counts.len() }
pub fn is_empty(&self) -> bool { self.counts.is_empty() }}
pub fn count_expanded( objects_dir: &Path, roots: Vec<Oid>, max_objects: ObjectCount, stall: Duration, kind: gix::hash::Kind,) -> Result<ExpandedPack, PackError> { let mut odb = odb_at(objects_dir, kind)?; odb.prevent_pack_unload(); odb.refresh_never();
let interrupt = Arc::new(AtomicBool::new(false)); let counter = Arc::new(AtomicUsize::new(0)); let progress = SharedCount(Arc::clone(&counter)); let oids: OidStream = Box::new(roots.into_iter().map(|oid| Ok(oid.object_id()))); let result = { let _watchdog = Watchdog::arm(interrupt.clone(), Arc::clone(&counter), max_objects, stall); output::count::objects( Interruptible { inner: odb.clone(), flag: Arc::clone(&interrupt), }, oids, &progress, &interrupt, output::count::objects::Options { thread_limit: knot_resource::gix_thread_limit() .map(knot_resource::ThreadCount::get), chunk_size: 50, input_object_expansion: output::count::objects::ObjectExpansion::TreeContents, }, ) }; let over_limit = counter.load(Ordering::Relaxed) > max_objects.get(); let timed_out = interrupt.load(Ordering::Relaxed); let counts = match result { Ok((counts, _)) => counts, Err(_) if over_limit => return Err(PackError::SelectionTooLarge), Err(_) if timed_out => return Err(PackError::SelectionTimeout), Err(error) => return Err(PackError::Pack(error.to_string())), }; if counts.len() > max_objects.get() { return Err(PackError::SelectionTooLarge); } if timed_out { return Err(PackError::SelectionTimeout); } Ok(ExpandedPack { counts, odb, kind })}
pub fn write_expanded(pack: ExpandedPack, out: &mut dyn Write) -> Result<(), PackError> { write_counts(pack.counts, pack.odb, None, out, pack.kind)}
fn write_counts( counts: Vec<output::Count>, odb: gix::odb::Handle, permitted_bases: Option<HashSet<ObjectId>>, out: &mut dyn Write, kind: gix::hash::Kind,) -> Result<(), PackError> { let num_entries = counts.len() as u32; let allow_thin_pack = permitted_bases.is_some(); let counted = output::entry::iter_from_counts( counts, odb.clone(), Box::new(Discard), output::entry::iter_from_counts::Options { thread_limit: knot_resource::gix_thread_limit().map(knot_resource::ThreadCount::get), mode: output::entry::iter_from_counts::Mode::PackCopyAndBaseObjects, allow_thin_pack, chunk_size: 50, version: Version::V2, }, );
let entries = gix::parallel::InOrderIter::from(counted).map( move |chunk| -> Result<Vec<output::Entry>, PackError> { let entries = chunk.map_err(|error| PackError::Pack(error.to_string()))?; entries .into_iter() .map(|entry| restrict_thin_base(&odb, permitted_bases.as_ref(), entry)) .collect() }, );
let mut writer = output::bytes::FromEntriesIter::new(entries, out, num_entries, Version::V2, kind); writer .try_fold((), |(), written| written.map(|_| ())) .map_err(|error| PackError::Pack(error.to_string()))?; Ok(())}
struct SharedCount(Arc<AtomicUsize>);
impl gix::progress::Count for SharedCount { fn set(&self, step: usize) { self.0.store(step, Ordering::Relaxed); }
fn step(&self) -> usize { self.0.load(Ordering::Relaxed) }
fn inc_by(&self, step: usize) { self.0.fetch_add(step, Ordering::Relaxed); }
fn counter(&self) -> Arc<AtomicUsize> { Arc::clone(&self.0) }}
#[derive(Debug)]struct Halted;
impl std::fmt::Display for Halted { fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { f.write_str("object enumeration halted by selection budget") }}
impl Error for Halted {}
#[derive(Clone)]struct Interruptible { inner: gix::odb::Handle, flag: Arc<AtomicBool>,}
impl gix_pack::Find for Interruptible { fn contains(&self, id: &gix::hash::oid) -> bool { self.inner.contains(id) }
fn try_find_cached<'a>( &self, id: &gix::hash::oid, buffer: &'a mut Vec<u8>, pack_cache: &mut dyn gix_pack::cache::DecodeEntry, ) -> Result< Option<(gix::objs::Data<'a>, Option<gix_pack::data::entry::Location>)>, gix::objs::find::Error, > { if self.flag.load(Ordering::Relaxed) { return Err(Box::new(Halted)); } self.inner.try_find_cached(id, buffer, pack_cache) }
fn location_by_oid( &self, id: &gix::hash::oid, buf: &mut Vec<u8>, ) -> Option<gix_pack::data::entry::Location> { self.inner.location_by_oid(id, buf) }
fn pack_offsets_and_oid( &self, pack_id: u32, ) -> Option<Vec<(gix_pack::data::Offset, gix::hash::ObjectId)>> { self.inner.pack_offsets_and_oid(pack_id) }
fn entry_by_location( &self, location: &gix_pack::data::entry::Location, ) -> Option<gix_pack::find::Entry> { self.inner.entry_by_location(location) }}
struct Watchdog { idle: Option<mpsc::Sender<()>>, watch: Option<std::thread::JoinHandle<()>>,}
impl Watchdog { fn arm( flag: Arc<AtomicBool>, counter: Arc<AtomicUsize>, max_objects: ObjectCount, stall: Duration, ) -> Self { let (idle, wake) = mpsc::channel::<()>(); let watch = std::thread::spawn(move || { let poll = Duration::from_millis(25).min(stall); let mut last = counter.load(Ordering::Relaxed); let mut since = Instant::now(); loop { let now = counter.load(Ordering::Relaxed); if now != last { last = now; since = Instant::now(); } if since.elapsed() >= stall || now > max_objects.get() { flag.store(true, Ordering::Relaxed); return; } if !matches!( wake.recv_timeout(poll), Err(mpsc::RecvTimeoutError::Timeout) ) { return; } } }); Self { idle: Some(idle), watch: Some(watch), } }}
impl Drop for Watchdog { fn drop(&mut self) { self.idle.take(); if let Some(watch) = self.watch.take() { let _ = watch.join(); } }}
fn restrict_thin_base( odb: &gix::odb::Handle, permitted: Option<&HashSet<ObjectId>>, entry: output::Entry,) -> Result<output::Entry, PackError> { let base = match &entry.kind { output::entry::Kind::DeltaOid { id } => *id, _ => return Ok(entry), }; if permitted.is_some_and(|bases| bases.contains(&base)) { return Ok(entry); } let mut buf = Vec::new(); let object = odb .find(&entry.id, &mut buf) .map_err(|error| PackError::Pack(error.to_string()))?; let count = output::Count::from_data(entry.id, None); output::Entry::from_data(&count, &object).map_err(|error| PackError::Pack(error.to_string()))}
pub fn index_pack( objects_dir: &Path, pack: &[u8], limits: &PackLimits, kind: gix::hash::Kind,) -> Result<(), PackError> { if pack.is_empty() { return Ok(()); } if !pack.starts_with(b"PACK") { return Err(PackError::Pack( "packfile is missing its PACK signature".to_string(), )); } std::fs::create_dir_all(objects_dir)?; let mut tmp = tempfile::NamedTempFile::new_in(objects_dir)?; tmp.write_all(pack)?; tmp.flush()?; let file = gix_pack::data::File::at(tmp.path(), kind) .map_err(|error| PackError::Pack(error.to_string()))?; index_pack_bounded(objects_dir, &file, limits, kind)}
pub(crate) fn index_pack_bounded( objects_dir: &Path, pack: &gix_pack::data::File, limits: &PackLimits, kind: gix::hash::Kind,) -> Result<(), PackError> { if pack.data_len() < 12 + kind.len_in_bytes() { return Ok(()); } let thin = meter::meter_file(pack, limits, kind)?;
let pack_dir = objects_dir.join("pack"); std::fs::create_dir_all(&pack_dir)?;
inline_and_index(objects_dir, pack, &pack_dir, limits, kind, thin).or_else(|_| { let bytes = std::fs::read(pack.path())?; resolve::resolve(objects_dir, &bytes, limits, kind) })}
fn inline_and_index( objects_dir: &Path, pack: &gix_pack::data::File, pack_dir: &Path, limits: &PackLimits, kind: gix::hash::Kind, thin: bool,) -> Result<(), PackError> { let (pack_path, pack_hash) = if thin { let pack_hash = inline_thin_bases(objects_dir, pack, pack_dir, kind)?; ( pack_dir.join(format!("pack-{}.pack", pack_hash.to_hex())), pack_hash, ) } else { let pack_hash = pack.checksum(); let pack_path = pack_dir.join(format!("pack-{}.pack", pack_hash.to_hex())); let source = pack.path().to_owned(); persist_atomic(&pack_path, |writer| { std::io::copy(&mut std::fs::File::open(&source)?, writer)?; Ok(()) })?; (pack_path, pack_hash) }; let pack_cleanup = RemoveOnDrop::arm(&pack_path);
let nodes = scan_offsets(&pack_path, kind)?; let spool = spool_pack( &pack_path, nodes, kind, knot_resource::ingest_base_budget(), limits.max_object_bytes, None, )?; let (_present, idx_cleanup) = persist_index(pack_dir, &pack_hash, spool, kind)?; pack_cleanup.disarm(); idx_cleanup.disarm(); Ok(())}
fn inline_thin_bases( objects_dir: &Path, pack: &gix_pack::data::File, pack_dir: &Path, kind: gix::hash::Kind,) -> Result<ObjectId, PackError> { let odb = odb_at(objects_dir, kind)?; let staged = tempfile::NamedTempFile::new_in(pack_dir)?; let writer = std::fs::OpenOptions::new() .read(true) .write(true) .open(staged.path())?; let reader = std::io::BufReader::new(std::fs::File::open(pack.path())?); let entries = gix_pack::data::input::BytesToEntriesIter::new_from_header( reader, gix_pack::data::input::Mode::Verify, gix_pack::data::input::EntryDataMode::KeepAndCrc32, kind, ) .map_err(|error| PackError::Pack(error.to_string()))?; let version = entries.version(); let lookup = gix_pack::data::input::LookupRefDeltaObjectsIter::new(entries, odb); let mut sink = gix_pack::data::input::EntriesToBytesIter::new(lookup, writer, version, kind); sink.try_for_each(|entry| { entry .map(|_| ()) .map_err(|error| PackError::Pack(error.to_string())) })?; let pack_hash = sink .digest() .ok_or_else(|| PackError::Pack("resolved pack has no trailer".to_string()))?; drop(sink);
let pack_path = pack_dir.join(format!("pack-{}.pack", pack_hash.to_hex())); staged .persist(&pack_path) .map_err(|error| PackError::Pack(error.to_string()))?; Ok(pack_hash)}
fn scan_offsets(pack_path: &Path, kind: gix::hash::Kind) -> Result<Vec<Node>, PackError> { let reader = std::io::BufReader::new(std::fs::File::open(pack_path)?); let mut entries = gix_pack::data::input::BytesToEntriesIter::new_from_header( reader, gix_pack::data::input::Mode::Verify, gix_pack::data::input::EntryDataMode::Crc32, kind, ) .map_err(|error| PackError::Pack(error.to_string()))?; let mut nodes: Vec<Node> = Vec::new(); entries.try_for_each(|entry| -> Result<(), PackError> { let entry = entry.map_err(|error| PackError::Pack(error.to_string()))?; nodes.push(Node { offset: PackOffset::new(entry.pack_offset), crc32: Crc32::new( entry .crc32 .ok_or_else(|| PackError::Pack("entry crc32 not computed".to_string()))?, ), }); Ok(()) })?; Ok(nodes)}
pub(crate) struct PresentSet(gix_pack::index::File);
impl PresentSet { fn open(idx_path: &Path, kind: gix::hash::Kind) -> Result<Self, PackError> { gix_pack::index::File::at(idx_path, kind) .map(Self) .map_err(|error| PackError::Pack(error.to_string())) }
pub(crate) fn contains(&self, oid: &Oid) -> bool { self.0.lookup(oid.object_id()).is_some() }
fn sorted_offsets(&self) -> Vec<PackOffset> { self.0 .sorted_offsets() .into_iter() .map(PackOffset::new) .collect() }}
pub(crate) struct FreshClosure { pub self_contained: bool, pub present: PresentSet,}
#[derive(Clone)]struct Node { offset: PackOffset, crc32: Crc32,}
const PRESENT: u8 = 1;const REFERENCED: u8 = 2;
struct Connectivity { objects: scc::HashMap<Oid, u8>, empty_tree: Oid,}
impl Connectivity { fn mark_present(&self, oid: Oid) { self.objects .entry_sync(oid) .and_modify(|flags| *flags |= PRESENT) .or_insert(PRESENT); }
fn check(&self, reference: Oid) { if reference != self.empty_tree { self.objects .entry_sync(reference) .and_modify(|flags| *flags |= REFERENCED) .or_insert(REFERENCED); } }
fn self_contained(&self) -> bool { self.objects .any_sync(|_, flags| *flags & REFERENCED != 0 && *flags & PRESENT == 0) .is_none() }}
enum Scan { Thin, Limit(PackLimit), Pack(String),}
const INGEST_BYTES_PER_OBJECT: u64 = 128;
fn fits_in_memory(num_objects: ObjectCount) -> bool { knot_resource::ingest_admits(knot_resource::PayloadBytes::new( (num_objects.get() as u64).saturating_mul(INGEST_BYTES_PER_OBJECT), ))}
pub(crate) fn admit_ingest( pack: &gix_pack::data::File, kind: gix::hash::Kind,) -> Result<(), PackError> { if pack.data_len() < 12 + kind.len_in_bytes() { return Ok(()); } if fits_in_memory(ObjectCount::from(pack.num_objects())) { Ok(()) } else { Err(PackError::InsufficientMemory) }}
struct RemoveOnDrop(Option<PathBuf>);
impl RemoveOnDrop { fn arm(path: &Path) -> Self { Self(Some(path.to_owned())) }
fn disarm(mut self) { self.0 = None; }}
impl Drop for RemoveOnDrop { fn drop(&mut self) { if let Some(path) = self.0.take() { let _ = std::fs::remove_file(path); } }}
pub(crate) fn ingest_and_close( objects_dir: &Path, pack: &gix_pack::data::File, limits: &PackLimits, kind: gix::hash::Kind, base_budget: Option<usize>, force_external: bool,) -> Result<Option<FreshClosure>, PackError> { let hash_len = kind.len_in_bytes(); if pack.data_len() < 12 + hash_len { return Ok(None); } if ObjectCount::from(pack.num_objects()) > limits.max_objects { return Err(PackError::LimitExceeded(PackLimit::Objects)); }
let reader = std::io::BufReader::new(std::fs::File::open(pack.path())?); let mut entries = gix_pack::data::input::BytesToEntriesIter::new_from_header( reader, gix_pack::data::input::Mode::Verify, gix_pack::data::input::EntryDataMode::Crc32, kind, ) .map_err(|error| PackError::Pack(error.to_string()))?; let mut nodes: Vec<Node> = Vec::with_capacity(pack.num_objects() as usize); let mut base_of: HashMap<PackOffset, PackOffset> = HashMap::new(); let mut total_decompressed: u64 = 0; let mut max_object: u64 = 0; let scan = entries.try_for_each(|entry| -> Result<(), Scan> { let entry = entry.map_err(|error| Scan::Pack(error.to_string()))?; match entry.header { gix_pack::data::entry::Header::RefDelta { .. } => return Err(Scan::Thin), gix_pack::data::entry::Header::OfsDelta { base_distance } => { let pack_offset = PackOffset::new(entry.pack_offset); let base = pack_offset .checked_sub_distance(base_distance) .ok_or_else(|| Scan::Pack("ofs-delta base out of range".to_string()))?; base_of.insert(pack_offset, base); } _ => {} } if limits.max_object_bytes.exceeded_by(entry.decompressed_size) { return Err(Scan::Limit(PackLimit::ObjectBytes)); } max_object = max_object.max(entry.decompressed_size); total_decompressed = total_decompressed .checked_add(entry.decompressed_size) .ok_or_else(|| Scan::Pack("decompressed size overflow".to_string()))?; if limits.max_total_bytes.exceeded_by(total_decompressed) { return Err(Scan::Limit(PackLimit::TotalBytes)); } nodes.push(Node { offset: PackOffset::new(entry.pack_offset), crc32: Crc32::new( entry .crc32 .ok_or_else(|| Scan::Pack("entry crc32 not computed".to_string()))?, ), }); Ok(()) }); match scan { Ok(()) => {} Err(Scan::Thin) => return Ok(None), Err(Scan::Limit(limit)) => return Err(PackError::LimitExceeded(limit)), Err(Scan::Pack(message)) => return Err(PackError::Pack(message)), } check_depth(&base_of, limits.max_delta_depth)?; drop(base_of); let entry_count = nodes.len();
let base_cache = knot_resource::ingest_base_budget().unwrap_or(0) as u64; let working_set = (entry_count as u64) .saturating_mul(INGEST_BYTES_PER_OBJECT) .saturating_add(base_cache) .saturating_add(max_object); if !knot_resource::ingest_admits_churn( knot_resource::WorkingSetBytes::new(working_set), knot_resource::ChurnBytes::new(total_decompressed), ) { return Err(PackError::InsufficientMemory); }
let pack_hash = pack.checksum(); let pack_dir = objects_dir.join("pack"); std::fs::create_dir_all(&pack_dir)?; let stem = format!("pack-{}", pack_hash.to_hex()); let pack_path = pack_dir.join(format!("{stem}.pack")); let source = pack.path().to_owned(); persist_atomic(&pack_path, |writer| { let mut reader = std::fs::File::open(&source)?; std::io::copy(&mut reader, writer)?; Ok(()) })?; let pack_cleanup = RemoveOnDrop::arm(&pack_path);
let externalize = force_external || knot_resource::externalize_connectivity(knot_resource::ConnectivityObjects::new( entry_count as u64, )); let connectivity = (!externalize).then(|| Connectivity { objects: scc::HashMap::with_capacity(entry_count), empty_tree: Oid::from(ObjectId::empty_tree(kind)), }); let spool = spool_pack( &pack_path, nodes, kind, base_budget, limits.max_object_bytes, connectivity.as_ref(), )?; let self_contained_in_ram = connectivity.as_ref().map(Connectivity::self_contained); drop(connectivity); let (present, idx_cleanup) = persist_index(&pack_dir, &pack_hash, spool, kind)?;
let self_contained = match self_contained_in_ram { Some(value) => value, None => verify_connectivity_retraverse( &pack_path, &present, kind, base_budget, limits.max_object_bytes, )?, };
pack_cleanup.disarm(); idx_cleanup.disarm(); Ok(Some(FreshClosure { self_contained, present, }))}
fn spool_pack( pack_path: &Path, nodes: Vec<Node>, kind: gix::hash::Kind, base_budget: Option<usize>, max_object_bytes: MaxObjectBytes, connectivity: Option<&Connectivity>,) -> Result<crate::idxwrite::Spool, PackError> { let interrupt = AtomicBool::new(false); let tree = gix_pack::cache::delta::Tree::from_offsets_in_pack( pack_path, nodes.into_iter(), &|node: &Node| node.offset.get(), &|_id| None, &mut Discard, &interrupt, kind, ) .map_err(|error| PackError::Pack(error.to_string()))?; let stored = gix_pack::data::File::at(pack_path, kind) .map_err(|error| PackError::Pack(error.to_string()))?; let spool = crate::idxwrite::Spool::new(kind)?; run_ingest_traverse( tree, &stored, &interrupt, base_budget, max_object_bytes, kind, |node: &mut Node, _progress, context| harvest(node, context, kind, connectivity, &spool), )?; Ok(spool)}
fn persist_index( pack_dir: &Path, pack_hash: &ObjectId, spool: crate::idxwrite::Spool, kind: gix::hash::Kind,) -> Result<(PresentSet, RemoveOnDrop), PackError> { let idx_path = pack_dir.join(format!("pack-{}.idx", pack_hash.to_hex())); persist_atomic(&idx_path, |writer| { crate::idxwrite::write_v2_index(writer, &spool, pack_hash, kind).map(|_| ()) })?; let idx_cleanup = RemoveOnDrop::arm(&idx_path); drop(spool); let present = PresentSet::open(&idx_path, kind)?; Ok((present, idx_cleanup))}
// he wishes he was on the farm alreadyfn harvest( node: &Node, context: gix_pack::cache::delta::traverse::Context<'_>, kind: gix::hash::Kind, connectivity: Option<&Connectivity>, spool: &crate::idxwrite::Spool,) -> Result<(), PackError> { let id = gix::objs::compute_hash(kind, context.object_kind, context.decompressed) .map_err(|error| PackError::Pack(error.to_string()))?; spool.push(id, node.crc32, node.offset)?; if let Some(connectivity) = connectivity { connectivity.mark_present(Oid::from(id)); parse_references( context.object_kind, context.decompressed, kind, &mut |reference| connectivity.check(reference), )?; } Ok(())}
fn parse_references( object_kind: gix::object::Kind, bytes: &[u8], kind: gix::hash::Kind, check: &mut dyn FnMut(Oid),) -> Result<(), PackError> { match object_kind { gix::object::Kind::Blob => Ok(()), gix::object::Kind::Tree => harvest_tree(bytes, kind, check), gix::object::Kind::Commit => harvest_commit(bytes, kind, check), gix::object::Kind::Tag => harvest_tag(bytes, kind, check), }}
fn harvest_tree( bytes: &[u8], kind: gix::hash::Kind, check: &mut dyn FnMut(Oid),) -> Result<(), PackError> { gix::objs::TreeRefIter::from_bytes(bytes, kind).try_for_each(|entry| { let entry = entry.map_err(|error| PackError::Pack(error.to_string()))?; if !entry.mode.is_commit() { check(Oid::from(entry.oid.to_owned())); } Ok(()) })}
fn harvest_commit( bytes: &[u8], kind: gix::hash::Kind, check: &mut dyn FnMut(Oid),) -> Result<(), PackError> { let mut iter = gix::objs::CommitRefIter::from_bytes(bytes, kind); let tree = iter .tree_id() .map_err(|error| PackError::Pack(error.to_string()))?; std::iter::once(tree) .chain(iter.parent_ids()) .for_each(|oid| check(Oid::from(oid))); Ok(())}
fn harvest_tag( bytes: &[u8], kind: gix::hash::Kind, check: &mut dyn FnMut(Oid),) -> Result<(), PackError> { let target = gix::objs::TagRefIter::from_bytes(bytes, kind) .target_id() .map_err(|error| PackError::Pack(error.to_string()))?; check(Oid::from(target)); Ok(())}
fn verify_connectivity_retraverse( pack_path: &Path, present: &PresentSet, kind: gix::hash::Kind, base_budget: Option<usize>, max_object_bytes: MaxObjectBytes,) -> Result<bool, PackError> { let interrupt = AtomicBool::new(false); let tree = gix_pack::cache::delta::Tree::from_offsets_in_pack( pack_path, present.sorted_offsets().into_iter(), &|offset: &PackOffset| offset.get(), &|_id| None, &mut Discard, &interrupt, kind, ) .map_err(|error| PackError::Pack(error.to_string()))?; let stored = gix_pack::data::File::at(pack_path, kind) .map_err(|error| PackError::Pack(error.to_string()))?; let empty_tree = Oid::from(ObjectId::empty_tree(kind)); let missing = AtomicBool::new(false); run_ingest_traverse( tree, &stored, &interrupt, base_budget, max_object_bytes, kind, |_offset: &mut PackOffset, _progress, context| { parse_references( context.object_kind, context.decompressed, kind, &mut |reference| { if reference != empty_tree && !present.contains(&reference) { missing.store(true, Ordering::Relaxed); } }, ) }, )?; Ok(!missing.load(Ordering::Relaxed))}
fn run_ingest_traverse<T, H>( tree: gix_pack::cache::delta::Tree<T>, stored: &gix_pack::data::File, interrupt: &AtomicBool, base_budget: Option<usize>, max_object_bytes: MaxObjectBytes, kind: gix::hash::Kind, harvest: H,) -> Result<(), PackError>where T: Send + Sync + Clone, H: FnMut( &mut T, &dyn gix::progress::Progress, gix_pack::cache::delta::traverse::Context<'_>, ) -> Result<(), PackError> + Send + Clone,{ let base_spill = match base_budget { Some(budget) => Some(Arc::new(gix_pack::cache::delta::traverse::BaseSpill::new( tempfile::tempfile()?, budget, ))), None => None, }; tree.traverse( |range: gix_pack::data::EntryRange, source: &gix_pack::data::File, buf: &mut Vec<u8>| { source.read_into(range, buf) }, stored, stored.pack_end() as u64, harvest, gix_pack::cache::delta::traverse::Options { object_progress: Box::new(Discard), size_progress: &mut Discard, thread_limit: Some(knot_resource::ingest_thread_limit()), should_interrupt: interrupt, object_hash: kind, base_spill, collect_items: false, max_object_bytes: Some(max_object_bytes.get()), }, ) .map(|_| ()) .map_err(|error| PackError::Pack(error.to_string()))}
fn persist_atomic( path: &Path, write: impl FnOnce(&mut dyn Write) -> Result<(), PackError>,) -> Result<(), PackError> { let dir = path .parent() .ok_or_else(|| PackError::Pack("object path has no parent".to_string()))?; let mut tmp = tempfile::NamedTempFile::new_in(dir)?; let mut writer = std::io::BufWriter::new(tmp.as_file_mut()); write(&mut writer)?; writer.flush()?; drop(writer); tmp.as_file().sync_all()?; tmp.persist(path) .map_err(|error| PackError::Pack(error.to_string()))?; std::fs::File::open(dir)?.sync_all()?; Ok(())}
#[cfg(test)]mod tests { use gix_pack::Find;
use super::*;
#[test] fn interruptible_find_errors_once_the_flag_is_set() { let dir = tempfile::tempdir().unwrap(); let flag = Arc::new(AtomicBool::new(false)); let find = Interruptible { inner: gix::odb::at(dir.path()).unwrap(), flag: Arc::clone(&flag), }; let absent = Oid::null().object_id(); let mut buf = Vec::new();
assert!( find.try_find(&absent, &mut buf).unwrap().is_none(), "before budget trips, lookup of an absent object is a plain miss" );
flag.store(true, Ordering::Relaxed); assert!( find.try_find(&absent, &mut buf).is_err(), "once tripped, every decode errors, which is what aborts a breadthfirst tree walk \ mid-closure instead of waiting for the next commit-root boundary" ); }}