alice.pds.demo.boltless.dev
This repository has no description
did:plc:3fwecdnvtcscjnrx2p4n7a…
1use std::collections::{HashMap, HashSet};
2use std::error::Error;
3use std::io::Write;
4use std::path::{Path, PathBuf};
5use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering};
6use std::sync::{Arc, mpsc};
7use std::time::{Duration, Instant};
8
9use gix::ObjectId;
10use gix::prelude::FindExt;
11use gix::progress::Discard;
12use gix_pack::data::{Version, output};
13use knot_types::{ObjectCount, Oid};
14
15use crate::error::{PackError, PackLimit};
16use crate::ids::{Crc32, MaxObjectBytes, PackOffset};
17use crate::meter::{self, PackLimits, check_depth};
18use crate::resolve;
19
20type OidStream = Box<dyn Iterator<Item = Result<ObjectId, Box<dyn Error + Send + Sync>>> + Send>;
21
22fn odb_at(objects_dir: &Path, kind: gix::hash::Kind) -> Result<gix::odb::Handle, PackError> {
23 gix::odb::at_opts(
24 objects_dir,
25 std::iter::empty(),
26 gix::odb::store::init::Options {
27 object_hash: kind,
28 ..Default::default()
29 },
30 )
31 .map_err(|error| PackError::Pack(error.to_string()))
32}
33
34pub fn write_pack(
35 objects_dir: &Path,
36 oids: Vec<Oid>,
37 thin_bases: Option<&HashSet<Oid>>,
38 out: &mut dyn Write,
39 kind: gix::hash::Kind,
40) -> Result<(), PackError> {
41 let mut odb = odb_at(objects_dir, kind)?;
42 odb.prevent_pack_unload();
43 odb.refresh_never();
44
45 let interrupt = AtomicBool::new(false);
46 let permitted_bases: Option<HashSet<ObjectId>> =
47 thin_bases.map(|bases| bases.iter().map(|oid| oid.object_id()).collect());
48 let oids: OidStream = Box::new(oids.into_iter().map(|oid| Ok(oid.object_id())));
49
50 let (counts, _) = output::count::objects(
51 odb.clone(),
52 oids,
53 &Discard,
54 &interrupt,
55 output::count::objects::Options {
56 thread_limit: knot_resource::gix_thread_limit().map(knot_resource::ThreadCount::get),
57 chunk_size: 50,
58 input_object_expansion: output::count::objects::ObjectExpansion::AsIs,
59 },
60 )
61 .map_err(|error| PackError::Pack(error.to_string()))?;
62
63 write_counts(counts, odb, permitted_bases, out, kind)
64}
65
66pub struct ExpandedPack {
67 counts: Vec<output::Count>,
68 odb: gix::odb::Handle,
69 kind: gix::hash::Kind,
70}
71
72impl ExpandedPack {
73 pub fn len(&self) -> usize {
74 self.counts.len()
75 }
76
77 pub fn is_empty(&self) -> bool {
78 self.counts.is_empty()
79 }
80}
81
82pub fn count_expanded(
83 objects_dir: &Path,
84 roots: Vec<Oid>,
85 max_objects: ObjectCount,
86 stall: Duration,
87 kind: gix::hash::Kind,
88) -> Result<ExpandedPack, PackError> {
89 let mut odb = odb_at(objects_dir, kind)?;
90 odb.prevent_pack_unload();
91 odb.refresh_never();
92
93 let interrupt = Arc::new(AtomicBool::new(false));
94 let counter = Arc::new(AtomicUsize::new(0));
95 let progress = SharedCount(Arc::clone(&counter));
96 let oids: OidStream = Box::new(roots.into_iter().map(|oid| Ok(oid.object_id())));
97 let result = {
98 let _watchdog = Watchdog::arm(interrupt.clone(), Arc::clone(&counter), max_objects, stall);
99 output::count::objects(
100 Interruptible {
101 inner: odb.clone(),
102 flag: Arc::clone(&interrupt),
103 },
104 oids,
105 &progress,
106 &interrupt,
107 output::count::objects::Options {
108 thread_limit: knot_resource::gix_thread_limit()
109 .map(knot_resource::ThreadCount::get),
110 chunk_size: 50,
111 input_object_expansion: output::count::objects::ObjectExpansion::TreeContents,
112 },
113 )
114 };
115 let over_limit = counter.load(Ordering::Relaxed) > max_objects.get();
116 let timed_out = interrupt.load(Ordering::Relaxed);
117 let counts = match result {
118 Ok((counts, _)) => counts,
119 Err(_) if over_limit => return Err(PackError::SelectionTooLarge),
120 Err(_) if timed_out => return Err(PackError::SelectionTimeout),
121 Err(error) => return Err(PackError::Pack(error.to_string())),
122 };
123 if counts.len() > max_objects.get() {
124 return Err(PackError::SelectionTooLarge);
125 }
126 if timed_out {
127 return Err(PackError::SelectionTimeout);
128 }
129 Ok(ExpandedPack { counts, odb, kind })
130}
131
132pub fn write_expanded(pack: ExpandedPack, out: &mut dyn Write) -> Result<(), PackError> {
133 write_counts(pack.counts, pack.odb, None, out, pack.kind)
134}
135
136fn write_counts(
137 counts: Vec<output::Count>,
138 odb: gix::odb::Handle,
139 permitted_bases: Option<HashSet<ObjectId>>,
140 out: &mut dyn Write,
141 kind: gix::hash::Kind,
142) -> Result<(), PackError> {
143 let num_entries = counts.len() as u32;
144 let allow_thin_pack = permitted_bases.is_some();
145 let counted = output::entry::iter_from_counts(
146 counts,
147 odb.clone(),
148 Box::new(Discard),
149 output::entry::iter_from_counts::Options {
150 thread_limit: knot_resource::gix_thread_limit().map(knot_resource::ThreadCount::get),
151 mode: output::entry::iter_from_counts::Mode::PackCopyAndBaseObjects,
152 allow_thin_pack,
153 chunk_size: 50,
154 version: Version::V2,
155 },
156 );
157
158 let entries = gix::parallel::InOrderIter::from(counted).map(
159 move |chunk| -> Result<Vec<output::Entry>, PackError> {
160 let entries = chunk.map_err(|error| PackError::Pack(error.to_string()))?;
161 entries
162 .into_iter()
163 .map(|entry| restrict_thin_base(&odb, permitted_bases.as_ref(), entry))
164 .collect()
165 },
166 );
167
168 let mut writer =
169 output::bytes::FromEntriesIter::new(entries, out, num_entries, Version::V2, kind);
170 writer
171 .try_fold((), |(), written| written.map(|_| ()))
172 .map_err(|error| PackError::Pack(error.to_string()))?;
173 Ok(())
174}
175
176struct SharedCount(Arc<AtomicUsize>);
177
178impl gix::progress::Count for SharedCount {
179 fn set(&self, step: usize) {
180 self.0.store(step, Ordering::Relaxed);
181 }
182
183 fn step(&self) -> usize {
184 self.0.load(Ordering::Relaxed)
185 }
186
187 fn inc_by(&self, step: usize) {
188 self.0.fetch_add(step, Ordering::Relaxed);
189 }
190
191 fn counter(&self) -> Arc<AtomicUsize> {
192 Arc::clone(&self.0)
193 }
194}
195
196#[derive(Debug)]
197struct Halted;
198
199impl std::fmt::Display for Halted {
200 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
201 f.write_str("object enumeration halted by selection budget")
202 }
203}
204
205impl Error for Halted {}
206
207#[derive(Clone)]
208struct Interruptible {
209 inner: gix::odb::Handle,
210 flag: Arc<AtomicBool>,
211}
212
213impl gix_pack::Find for Interruptible {
214 fn contains(&self, id: &gix::hash::oid) -> bool {
215 self.inner.contains(id)
216 }
217
218 fn try_find_cached<'a>(
219 &self,
220 id: &gix::hash::oid,
221 buffer: &'a mut Vec<u8>,
222 pack_cache: &mut dyn gix_pack::cache::DecodeEntry,
223 ) -> Result<
224 Option<(gix::objs::Data<'a>, Option<gix_pack::data::entry::Location>)>,
225 gix::objs::find::Error,
226 > {
227 if self.flag.load(Ordering::Relaxed) {
228 return Err(Box::new(Halted));
229 }
230 self.inner.try_find_cached(id, buffer, pack_cache)
231 }
232
233 fn location_by_oid(
234 &self,
235 id: &gix::hash::oid,
236 buf: &mut Vec<u8>,
237 ) -> Option<gix_pack::data::entry::Location> {
238 self.inner.location_by_oid(id, buf)
239 }
240
241 fn pack_offsets_and_oid(
242 &self,
243 pack_id: u32,
244 ) -> Option<Vec<(gix_pack::data::Offset, gix::hash::ObjectId)>> {
245 self.inner.pack_offsets_and_oid(pack_id)
246 }
247
248 fn entry_by_location(
249 &self,
250 location: &gix_pack::data::entry::Location,
251 ) -> Option<gix_pack::find::Entry> {
252 self.inner.entry_by_location(location)
253 }
254}
255
256struct Watchdog {
257 idle: Option<mpsc::Sender<()>>,
258 watch: Option<std::thread::JoinHandle<()>>,
259}
260
261impl Watchdog {
262 fn arm(
263 flag: Arc<AtomicBool>,
264 counter: Arc<AtomicUsize>,
265 max_objects: ObjectCount,
266 stall: Duration,
267 ) -> Self {
268 let (idle, wake) = mpsc::channel::<()>();
269 let watch = std::thread::spawn(move || {
270 let poll = Duration::from_millis(25).min(stall);
271 let mut last = counter.load(Ordering::Relaxed);
272 let mut since = Instant::now();
273 loop {
274 let now = counter.load(Ordering::Relaxed);
275 if now != last {
276 last = now;
277 since = Instant::now();
278 }
279 if since.elapsed() >= stall || now > max_objects.get() {
280 flag.store(true, Ordering::Relaxed);
281 return;
282 }
283 if !matches!(
284 wake.recv_timeout(poll),
285 Err(mpsc::RecvTimeoutError::Timeout)
286 ) {
287 return;
288 }
289 }
290 });
291 Self {
292 idle: Some(idle),
293 watch: Some(watch),
294 }
295 }
296}
297
298impl Drop for Watchdog {
299 fn drop(&mut self) {
300 self.idle.take();
301 if let Some(watch) = self.watch.take() {
302 let _ = watch.join();
303 }
304 }
305}
306
307fn restrict_thin_base(
308 odb: &gix::odb::Handle,
309 permitted: Option<&HashSet<ObjectId>>,
310 entry: output::Entry,
311) -> Result<output::Entry, PackError> {
312 let base = match &entry.kind {
313 output::entry::Kind::DeltaOid { id } => *id,
314 _ => return Ok(entry),
315 };
316 if permitted.is_some_and(|bases| bases.contains(&base)) {
317 return Ok(entry);
318 }
319 let mut buf = Vec::new();
320 let object = odb
321 .find(&entry.id, &mut buf)
322 .map_err(|error| PackError::Pack(error.to_string()))?;
323 let count = output::Count::from_data(entry.id, None);
324 output::Entry::from_data(&count, &object).map_err(|error| PackError::Pack(error.to_string()))
325}
326
327pub fn index_pack(
328 objects_dir: &Path,
329 pack: &[u8],
330 limits: &PackLimits,
331 kind: gix::hash::Kind,
332) -> Result<(), PackError> {
333 if pack.is_empty() {
334 return Ok(());
335 }
336 if !pack.starts_with(b"PACK") {
337 return Err(PackError::Pack(
338 "packfile is missing its PACK signature".to_string(),
339 ));
340 }
341 std::fs::create_dir_all(objects_dir)?;
342 let mut tmp = tempfile::NamedTempFile::new_in(objects_dir)?;
343 tmp.write_all(pack)?;
344 tmp.flush()?;
345 let file = gix_pack::data::File::at(tmp.path(), kind)
346 .map_err(|error| PackError::Pack(error.to_string()))?;
347 index_pack_bounded(objects_dir, &file, limits, kind)
348}
349
350pub(crate) fn index_pack_bounded(
351 objects_dir: &Path,
352 pack: &gix_pack::data::File,
353 limits: &PackLimits,
354 kind: gix::hash::Kind,
355) -> Result<(), PackError> {
356 if pack.data_len() < 12 + kind.len_in_bytes() {
357 return Ok(());
358 }
359 let thin = meter::meter_file(pack, limits, kind)?;
360
361 let pack_dir = objects_dir.join("pack");
362 std::fs::create_dir_all(&pack_dir)?;
363
364 inline_and_index(objects_dir, pack, &pack_dir, limits, kind, thin).or_else(|_| {
365 let bytes = std::fs::read(pack.path())?;
366 resolve::resolve(objects_dir, &bytes, limits, kind)
367 })
368}
369
370fn inline_and_index(
371 objects_dir: &Path,
372 pack: &gix_pack::data::File,
373 pack_dir: &Path,
374 limits: &PackLimits,
375 kind: gix::hash::Kind,
376 thin: bool,
377) -> Result<(), PackError> {
378 let (pack_path, pack_hash) = if thin {
379 let pack_hash = inline_thin_bases(objects_dir, pack, pack_dir, kind)?;
380 (
381 pack_dir.join(format!("pack-{}.pack", pack_hash.to_hex())),
382 pack_hash,
383 )
384 } else {
385 let pack_hash = pack.checksum();
386 let pack_path = pack_dir.join(format!("pack-{}.pack", pack_hash.to_hex()));
387 let source = pack.path().to_owned();
388 persist_atomic(&pack_path, |writer| {
389 std::io::copy(&mut std::fs::File::open(&source)?, writer)?;
390 Ok(())
391 })?;
392 (pack_path, pack_hash)
393 };
394 let pack_cleanup = RemoveOnDrop::arm(&pack_path);
395
396 let nodes = scan_offsets(&pack_path, kind)?;
397 let spool = spool_pack(
398 &pack_path,
399 nodes,
400 kind,
401 knot_resource::ingest_base_budget(),
402 limits.max_object_bytes,
403 None,
404 )?;
405 let (_present, idx_cleanup) = persist_index(pack_dir, &pack_hash, spool, kind)?;
406 pack_cleanup.disarm();
407 idx_cleanup.disarm();
408 Ok(())
409}
410
411fn inline_thin_bases(
412 objects_dir: &Path,
413 pack: &gix_pack::data::File,
414 pack_dir: &Path,
415 kind: gix::hash::Kind,
416) -> Result<ObjectId, PackError> {
417 let odb = odb_at(objects_dir, kind)?;
418 let staged = tempfile::NamedTempFile::new_in(pack_dir)?;
419 let writer = std::fs::OpenOptions::new()
420 .read(true)
421 .write(true)
422 .open(staged.path())?;
423 let reader = std::io::BufReader::new(std::fs::File::open(pack.path())?);
424 let entries = gix_pack::data::input::BytesToEntriesIter::new_from_header(
425 reader,
426 gix_pack::data::input::Mode::Verify,
427 gix_pack::data::input::EntryDataMode::KeepAndCrc32,
428 kind,
429 )
430 .map_err(|error| PackError::Pack(error.to_string()))?;
431 let version = entries.version();
432 let lookup = gix_pack::data::input::LookupRefDeltaObjectsIter::new(entries, odb);
433 let mut sink = gix_pack::data::input::EntriesToBytesIter::new(lookup, writer, version, kind);
434 sink.try_for_each(|entry| {
435 entry
436 .map(|_| ())
437 .map_err(|error| PackError::Pack(error.to_string()))
438 })?;
439 let pack_hash = sink
440 .digest()
441 .ok_or_else(|| PackError::Pack("resolved pack has no trailer".to_string()))?;
442 drop(sink);
443
444 let pack_path = pack_dir.join(format!("pack-{}.pack", pack_hash.to_hex()));
445 staged
446 .persist(&pack_path)
447 .map_err(|error| PackError::Pack(error.to_string()))?;
448 Ok(pack_hash)
449}
450
451fn scan_offsets(pack_path: &Path, kind: gix::hash::Kind) -> Result<Vec<Node>, PackError> {
452 let reader = std::io::BufReader::new(std::fs::File::open(pack_path)?);
453 let mut entries = gix_pack::data::input::BytesToEntriesIter::new_from_header(
454 reader,
455 gix_pack::data::input::Mode::Verify,
456 gix_pack::data::input::EntryDataMode::Crc32,
457 kind,
458 )
459 .map_err(|error| PackError::Pack(error.to_string()))?;
460 let mut nodes: Vec<Node> = Vec::new();
461 entries.try_for_each(|entry| -> Result<(), PackError> {
462 let entry = entry.map_err(|error| PackError::Pack(error.to_string()))?;
463 nodes.push(Node {
464 offset: PackOffset::new(entry.pack_offset),
465 crc32: Crc32::new(
466 entry
467 .crc32
468 .ok_or_else(|| PackError::Pack("entry crc32 not computed".to_string()))?,
469 ),
470 });
471 Ok(())
472 })?;
473 Ok(nodes)
474}
475
476pub(crate) struct PresentSet(gix_pack::index::File);
477
478impl PresentSet {
479 fn open(idx_path: &Path, kind: gix::hash::Kind) -> Result<Self, PackError> {
480 gix_pack::index::File::at(idx_path, kind)
481 .map(Self)
482 .map_err(|error| PackError::Pack(error.to_string()))
483 }
484
485 pub(crate) fn contains(&self, oid: &Oid) -> bool {
486 self.0.lookup(oid.object_id()).is_some()
487 }
488
489 fn sorted_offsets(&self) -> Vec<PackOffset> {
490 self.0
491 .sorted_offsets()
492 .into_iter()
493 .map(PackOffset::new)
494 .collect()
495 }
496}
497
498pub(crate) struct FreshClosure {
499 pub self_contained: bool,
500 pub present: PresentSet,
501}
502
503#[derive(Clone)]
504struct Node {
505 offset: PackOffset,
506 crc32: Crc32,
507}
508
509const PRESENT: u8 = 1;
510const REFERENCED: u8 = 2;
511
512struct Connectivity {
513 objects: scc::HashMap<Oid, u8>,
514 empty_tree: Oid,
515}
516
517impl Connectivity {
518 fn mark_present(&self, oid: Oid) {
519 self.objects
520 .entry_sync(oid)
521 .and_modify(|flags| *flags |= PRESENT)
522 .or_insert(PRESENT);
523 }
524
525 fn check(&self, reference: Oid) {
526 if reference != self.empty_tree {
527 self.objects
528 .entry_sync(reference)
529 .and_modify(|flags| *flags |= REFERENCED)
530 .or_insert(REFERENCED);
531 }
532 }
533
534 fn self_contained(&self) -> bool {
535 self.objects
536 .any_sync(|_, flags| *flags & REFERENCED != 0 && *flags & PRESENT == 0)
537 .is_none()
538 }
539}
540
541enum Scan {
542 Thin,
543 Limit(PackLimit),
544 Pack(String),
545}
546
547const INGEST_BYTES_PER_OBJECT: u64 = 128;
548
549fn fits_in_memory(num_objects: ObjectCount) -> bool {
550 knot_resource::ingest_admits(knot_resource::PayloadBytes::new(
551 (num_objects.get() as u64).saturating_mul(INGEST_BYTES_PER_OBJECT),
552 ))
553}
554
555pub(crate) fn admit_ingest(
556 pack: &gix_pack::data::File,
557 kind: gix::hash::Kind,
558) -> Result<(), PackError> {
559 if pack.data_len() < 12 + kind.len_in_bytes() {
560 return Ok(());
561 }
562 if fits_in_memory(ObjectCount::from(pack.num_objects())) {
563 Ok(())
564 } else {
565 Err(PackError::InsufficientMemory)
566 }
567}
568
569struct RemoveOnDrop(Option<PathBuf>);
570
571impl RemoveOnDrop {
572 fn arm(path: &Path) -> Self {
573 Self(Some(path.to_owned()))
574 }
575
576 fn disarm(mut self) {
577 self.0 = None;
578 }
579}
580
581impl Drop for RemoveOnDrop {
582 fn drop(&mut self) {
583 if let Some(path) = self.0.take() {
584 let _ = std::fs::remove_file(path);
585 }
586 }
587}
588
589pub(crate) fn ingest_and_close(
590 objects_dir: &Path,
591 pack: &gix_pack::data::File,
592 limits: &PackLimits,
593 kind: gix::hash::Kind,
594 base_budget: Option<usize>,
595 force_external: bool,
596) -> Result<Option<FreshClosure>, PackError> {
597 let hash_len = kind.len_in_bytes();
598 if pack.data_len() < 12 + hash_len {
599 return Ok(None);
600 }
601 if ObjectCount::from(pack.num_objects()) > limits.max_objects {
602 return Err(PackError::LimitExceeded(PackLimit::Objects));
603 }
604
605 let reader = std::io::BufReader::new(std::fs::File::open(pack.path())?);
606 let mut entries = gix_pack::data::input::BytesToEntriesIter::new_from_header(
607 reader,
608 gix_pack::data::input::Mode::Verify,
609 gix_pack::data::input::EntryDataMode::Crc32,
610 kind,
611 )
612 .map_err(|error| PackError::Pack(error.to_string()))?;
613 let mut nodes: Vec<Node> = Vec::with_capacity(pack.num_objects() as usize);
614 let mut base_of: HashMap<PackOffset, PackOffset> = HashMap::new();
615 let mut total_decompressed: u64 = 0;
616 let mut max_object: u64 = 0;
617 let scan = entries.try_for_each(|entry| -> Result<(), Scan> {
618 let entry = entry.map_err(|error| Scan::Pack(error.to_string()))?;
619 match entry.header {
620 gix_pack::data::entry::Header::RefDelta { .. } => return Err(Scan::Thin),
621 gix_pack::data::entry::Header::OfsDelta { base_distance } => {
622 let pack_offset = PackOffset::new(entry.pack_offset);
623 let base = pack_offset
624 .checked_sub_distance(base_distance)
625 .ok_or_else(|| Scan::Pack("ofs-delta base out of range".to_string()))?;
626 base_of.insert(pack_offset, base);
627 }
628 _ => {}
629 }
630 if limits.max_object_bytes.exceeded_by(entry.decompressed_size) {
631 return Err(Scan::Limit(PackLimit::ObjectBytes));
632 }
633 max_object = max_object.max(entry.decompressed_size);
634 total_decompressed = total_decompressed
635 .checked_add(entry.decompressed_size)
636 .ok_or_else(|| Scan::Pack("decompressed size overflow".to_string()))?;
637 if limits.max_total_bytes.exceeded_by(total_decompressed) {
638 return Err(Scan::Limit(PackLimit::TotalBytes));
639 }
640 nodes.push(Node {
641 offset: PackOffset::new(entry.pack_offset),
642 crc32: Crc32::new(
643 entry
644 .crc32
645 .ok_or_else(|| Scan::Pack("entry crc32 not computed".to_string()))?,
646 ),
647 });
648 Ok(())
649 });
650 match scan {
651 Ok(()) => {}
652 Err(Scan::Thin) => return Ok(None),
653 Err(Scan::Limit(limit)) => return Err(PackError::LimitExceeded(limit)),
654 Err(Scan::Pack(message)) => return Err(PackError::Pack(message)),
655 }
656 check_depth(&base_of, limits.max_delta_depth)?;
657 drop(base_of);
658 let entry_count = nodes.len();
659
660 let base_cache = knot_resource::ingest_base_budget().unwrap_or(0) as u64;
661 let working_set = (entry_count as u64)
662 .saturating_mul(INGEST_BYTES_PER_OBJECT)
663 .saturating_add(base_cache)
664 .saturating_add(max_object);
665 if !knot_resource::ingest_admits_churn(
666 knot_resource::WorkingSetBytes::new(working_set),
667 knot_resource::ChurnBytes::new(total_decompressed),
668 ) {
669 return Err(PackError::InsufficientMemory);
670 }
671
672 let pack_hash = pack.checksum();
673 let pack_dir = objects_dir.join("pack");
674 std::fs::create_dir_all(&pack_dir)?;
675 let stem = format!("pack-{}", pack_hash.to_hex());
676 let pack_path = pack_dir.join(format!("{stem}.pack"));
677 let source = pack.path().to_owned();
678 persist_atomic(&pack_path, |writer| {
679 let mut reader = std::fs::File::open(&source)?;
680 std::io::copy(&mut reader, writer)?;
681 Ok(())
682 })?;
683 let pack_cleanup = RemoveOnDrop::arm(&pack_path);
684
685 let externalize = force_external
686 || knot_resource::externalize_connectivity(knot_resource::ConnectivityObjects::new(
687 entry_count as u64,
688 ));
689 let connectivity = (!externalize).then(|| Connectivity {
690 objects: scc::HashMap::with_capacity(entry_count),
691 empty_tree: Oid::from(ObjectId::empty_tree(kind)),
692 });
693 let spool = spool_pack(
694 &pack_path,
695 nodes,
696 kind,
697 base_budget,
698 limits.max_object_bytes,
699 connectivity.as_ref(),
700 )?;
701 let self_contained_in_ram = connectivity.as_ref().map(Connectivity::self_contained);
702 drop(connectivity);
703 let (present, idx_cleanup) = persist_index(&pack_dir, &pack_hash, spool, kind)?;
704
705 let self_contained = match self_contained_in_ram {
706 Some(value) => value,
707 None => verify_connectivity_retraverse(
708 &pack_path,
709 &present,
710 kind,
711 base_budget,
712 limits.max_object_bytes,
713 )?,
714 };
715
716 pack_cleanup.disarm();
717 idx_cleanup.disarm();
718 Ok(Some(FreshClosure {
719 self_contained,
720 present,
721 }))
722}
723
724fn spool_pack(
725 pack_path: &Path,
726 nodes: Vec<Node>,
727 kind: gix::hash::Kind,
728 base_budget: Option<usize>,
729 max_object_bytes: MaxObjectBytes,
730 connectivity: Option<&Connectivity>,
731) -> Result<crate::idxwrite::Spool, PackError> {
732 let interrupt = AtomicBool::new(false);
733 let tree = gix_pack::cache::delta::Tree::from_offsets_in_pack(
734 pack_path,
735 nodes.into_iter(),
736 &|node: &Node| node.offset.get(),
737 &|_id| None,
738 &mut Discard,
739 &interrupt,
740 kind,
741 )
742 .map_err(|error| PackError::Pack(error.to_string()))?;
743 let stored = gix_pack::data::File::at(pack_path, kind)
744 .map_err(|error| PackError::Pack(error.to_string()))?;
745 let spool = crate::idxwrite::Spool::new(kind);
746 run_ingest_traverse(
747 tree,
748 &stored,
749 &interrupt,
750 base_budget,
751 max_object_bytes,
752 kind,
753 |node: &mut Node, _progress, context| harvest(node, context, kind, connectivity, &spool),
754 )?;
755 Ok(spool)
756}
757
758fn persist_index(
759 pack_dir: &Path,
760 pack_hash: &ObjectId,
761 spool: crate::idxwrite::Spool,
762 kind: gix::hash::Kind,
763) -> Result<(PresentSet, RemoveOnDrop), PackError> {
764 let idx_path = pack_dir.join(format!("pack-{}.idx", pack_hash.to_hex()));
765 persist_atomic(&idx_path, |writer| {
766 crate::idxwrite::write_v2_index(writer, &spool, pack_hash, kind).map(|_| ())
767 })?;
768 let idx_cleanup = RemoveOnDrop::arm(&idx_path);
769 drop(spool);
770 let present = PresentSet::open(&idx_path, kind)?;
771 Ok((present, idx_cleanup))
772}
773
774// he wishes he was on the farm already
775fn harvest(
776 node: &Node,
777 context: gix_pack::cache::delta::traverse::Context<'_>,
778 kind: gix::hash::Kind,
779 connectivity: Option<&Connectivity>,
780 spool: &crate::idxwrite::Spool,
781) -> Result<(), PackError> {
782 let id = gix::objs::compute_hash(kind, context.object_kind, context.decompressed)
783 .map_err(|error| PackError::Pack(error.to_string()))?;
784 spool.push(id, node.crc32, node.offset)?;
785 if let Some(connectivity) = connectivity {
786 connectivity.mark_present(Oid::from(id));
787 parse_references(
788 context.object_kind,
789 context.decompressed,
790 kind,
791 &mut |reference| connectivity.check(reference),
792 )?;
793 }
794 Ok(())
795}
796
797fn parse_references(
798 object_kind: gix::object::Kind,
799 bytes: &[u8],
800 kind: gix::hash::Kind,
801 check: &mut dyn FnMut(Oid),
802) -> Result<(), PackError> {
803 match object_kind {
804 gix::object::Kind::Blob => Ok(()),
805 gix::object::Kind::Tree => harvest_tree(bytes, kind, check),
806 gix::object::Kind::Commit => harvest_commit(bytes, kind, check),
807 gix::object::Kind::Tag => harvest_tag(bytes, kind, check),
808 }
809}
810
811fn harvest_tree(
812 bytes: &[u8],
813 kind: gix::hash::Kind,
814 check: &mut dyn FnMut(Oid),
815) -> Result<(), PackError> {
816 gix::objs::TreeRefIter::from_bytes(bytes, kind).try_for_each(|entry| {
817 let entry = entry.map_err(|error| PackError::Pack(error.to_string()))?;
818 if !entry.mode.is_commit() {
819 check(Oid::from(entry.oid.to_owned()));
820 }
821 Ok(())
822 })
823}
824
825fn harvest_commit(
826 bytes: &[u8],
827 kind: gix::hash::Kind,
828 check: &mut dyn FnMut(Oid),
829) -> Result<(), PackError> {
830 let mut iter = gix::objs::CommitRefIter::from_bytes(bytes, kind);
831 let tree = iter
832 .tree_id()
833 .map_err(|error| PackError::Pack(error.to_string()))?;
834 std::iter::once(tree)
835 .chain(iter.parent_ids())
836 .for_each(|oid| check(Oid::from(oid)));
837 Ok(())
838}
839
840fn harvest_tag(
841 bytes: &[u8],
842 kind: gix::hash::Kind,
843 check: &mut dyn FnMut(Oid),
844) -> Result<(), PackError> {
845 let target = gix::objs::TagRefIter::from_bytes(bytes, kind)
846 .target_id()
847 .map_err(|error| PackError::Pack(error.to_string()))?;
848 check(Oid::from(target));
849 Ok(())
850}
851
852fn verify_connectivity_retraverse(
853 pack_path: &Path,
854 present: &PresentSet,
855 kind: gix::hash::Kind,
856 base_budget: Option<usize>,
857 max_object_bytes: MaxObjectBytes,
858) -> Result<bool, PackError> {
859 let interrupt = AtomicBool::new(false);
860 let tree = gix_pack::cache::delta::Tree::from_offsets_in_pack(
861 pack_path,
862 present.sorted_offsets().into_iter(),
863 &|offset: &PackOffset| offset.get(),
864 &|_id| None,
865 &mut Discard,
866 &interrupt,
867 kind,
868 )
869 .map_err(|error| PackError::Pack(error.to_string()))?;
870 let stored = gix_pack::data::File::at(pack_path, kind)
871 .map_err(|error| PackError::Pack(error.to_string()))?;
872 let empty_tree = Oid::from(ObjectId::empty_tree(kind));
873 let missing = AtomicBool::new(false);
874 run_ingest_traverse(
875 tree,
876 &stored,
877 &interrupt,
878 base_budget,
879 max_object_bytes,
880 kind,
881 |_offset: &mut PackOffset, _progress, context| {
882 parse_references(
883 context.object_kind,
884 context.decompressed,
885 kind,
886 &mut |reference| {
887 if reference != empty_tree && !present.contains(&reference) {
888 missing.store(true, Ordering::Relaxed);
889 }
890 },
891 )
892 },
893 )?;
894 Ok(!missing.load(Ordering::Relaxed))
895}
896
897fn run_ingest_traverse<T, H>(
898 tree: gix_pack::cache::delta::Tree<T>,
899 stored: &gix_pack::data::File,
900 interrupt: &AtomicBool,
901 base_budget: Option<usize>,
902 max_object_bytes: MaxObjectBytes,
903 kind: gix::hash::Kind,
904 harvest: H,
905) -> Result<(), PackError>
906where
907 T: Send + Sync + Clone,
908 H: FnMut(
909 &mut T,
910 &dyn gix::progress::Progress,
911 gix_pack::cache::delta::traverse::Context<'_>,
912 ) -> Result<(), PackError>
913 + Send
914 + Clone,
915{
916 let base_spill = match base_budget {
917 Some(budget) => Some(Arc::new(gix_pack::cache::delta::traverse::BaseSpill::new(
918 tempfile::tempfile()?,
919 budget,
920 ))),
921 None => None,
922 };
923 tree.traverse(
924 |range: gix_pack::data::EntryRange, source: &gix_pack::data::File, buf: &mut Vec<u8>| {
925 source.read_into(range, buf)
926 },
927 stored,
928 stored.pack_end() as u64,
929 harvest,
930 gix_pack::cache::delta::traverse::Options {
931 object_progress: Box::new(Discard),
932 size_progress: &mut Discard,
933 thread_limit: Some(knot_resource::ingest_thread_limit()),
934 should_interrupt: interrupt,
935 object_hash: kind,
936 base_spill,
937 collect_items: false,
938 max_object_bytes: Some(max_object_bytes.get()),
939 },
940 )
941 .map(|_| ())
942 .map_err(|error| PackError::Pack(error.to_string()))
943}
944
945fn persist_atomic(
946 path: &Path,
947 write: impl FnOnce(&mut dyn Write) -> Result<(), PackError>,
948) -> Result<(), PackError> {
949 let dir = path
950 .parent()
951 .ok_or_else(|| PackError::Pack("object path has no parent".to_string()))?;
952 let mut tmp = tempfile::NamedTempFile::new_in(dir)?;
953 let mut writer = std::io::BufWriter::new(tmp.as_file_mut());
954 write(&mut writer)?;
955 writer.flush()?;
956 drop(writer);
957 tmp.as_file().sync_all()?;
958 tmp.persist(path)
959 .map_err(|error| PackError::Pack(error.to_string()))?;
960 std::fs::File::open(dir)?.sync_all()?;
961 Ok(())
962}
963
964#[cfg(test)]
965mod tests {
966 use gix_pack::Find;
967
968 use super::*;
969
970 #[test]
971 fn interruptible_find_errors_once_the_flag_is_set() {
972 let dir = tempfile::tempdir().unwrap();
973 let flag = Arc::new(AtomicBool::new(false));
974 let find = Interruptible {
975 inner: gix::odb::at(dir.path()).unwrap(),
976 flag: Arc::clone(&flag),
977 };
978 let absent = Oid::null().object_id();
979 let mut buf = Vec::new();
980
981 assert!(
982 find.try_find(&absent, &mut buf).unwrap().is_none(),
983 "before budget trips, lookup of an absent object is a plain miss"
984 );
985
986 flag.store(true, Ordering::Relaxed);
987 assert!(
988 find.try_find(&absent, &mut buf).is_err(),
989 "once tripped, every decode errors, which is what aborts a breadthfirst tree walk \
990 mid-closure instead of waiting for the next commit-root boundary"
991 );
992 }
993}