1#[cfg(not(unix))]
5use std::io::{Seek, SeekFrom};
6#[cfg(unix)]
7use std::os::unix::fs::FileExt;
8use std::{
9 fs::File,
10 io::{self, BufRead, Cursor, Read, Write},
11 path::{Path, PathBuf},
12 sync::{
13 Arc, Mutex, OnceLock,
14 atomic::{AtomicUsize, Ordering},
15 },
16 thread::{self, JoinHandle},
17 time::Instant,
18};
19
20use axum::Error;
21use bytes::Bytes;
22use dashmap::DashMap;
23use flate2::bufread::ZlibDecoder;
24use futures_util::{Stream, StreamExt};
25use tempfile::NamedTempFile;
26use threadpool::ThreadPool;
27use tokio::sync::mpsc::UnboundedSender;
28use uuid::Uuid;
29
30pub use crate::internal::pack::stats::PackStats;
31use crate::{
32 errors::GitError,
33 hash::{HashKind, ObjectHash, get_hash_kind, set_hash_kind},
34 internal::{
35 metadata::{EntryMeta, MetaAttached},
36 object::types::ObjectType,
37 pack::{
38 DEFAULT_TMP_DIR, Pack,
39 cache::{_Cache, Caches},
40 cache_object::{CacheObject, CacheObjectInfo, MemSizeRecorder},
41 channel_reader::StreamBufReader,
42 entry::Entry,
43 utils,
44 waitlist::Waitlist,
45 wrapper::Wrapper,
46 },
47 },
48 utils::{CountingReader, HashAlgorithm},
49 zstdelta,
50};
51
52struct CrcCountingReader<R> {
54 inner: R,
55 bytes_read: u64,
56 crc: Option<crc32fast::Hasher>,
57}
58
59struct HashingReader<R> {
60 inner: R,
61 hash: HashAlgorithm,
62}
63
64impl<R> HashingReader<R> {
65 fn new(inner: R) -> Self {
66 Self {
67 inner,
68 hash: HashAlgorithm::new(),
69 }
70 }
71
72 fn current_hash(&self) -> Result<ObjectHash, GitError> {
73 ObjectHash::from_bytes(&self.hash.clone().finalize())
74 .map_err(|e| GitError::InvalidPackFile(format!("Read index error: {e}")))
75 }
76}
77
78impl<R: Read> HashingReader<R> {
79 fn read_without_hash(&mut self, buf: &mut [u8]) -> io::Result<usize> {
80 self.inner.read(buf)
81 }
82
83 fn read_exact_without_hash(&mut self, buf: &mut [u8]) -> io::Result<()> {
84 self.inner.read_exact(buf)
85 }
86}
87
88impl<R: Read> Read for HashingReader<R> {
89 fn read(&mut self, buf: &mut [u8]) -> io::Result<usize> {
90 let n = self.inner.read(buf)?;
91 self.hash.update(&buf[..n]);
92 Ok(n)
93 }
94}
95
96impl<R: Read> Read for CrcCountingReader<R> {
97 fn read(&mut self, buf: &mut [u8]) -> io::Result<usize> {
98 let n = self.inner.read(buf)?;
99 self.bytes_read += n as u64;
100 if let Some(crc) = &mut self.crc {
101 crc.update(&buf[..n]);
102 }
103 Ok(n)
104 }
105}
106impl<R: BufRead> BufRead for CrcCountingReader<R> {
107 fn fill_buf(&mut self) -> io::Result<&[u8]> {
108 self.inner.fill_buf()
109 }
110 fn consume(&mut self, amt: usize) {
111 if let Some(crc) = &mut self.crc {
112 let buf = self.inner.fill_buf().unwrap_or(&[]);
113 crc.update(&buf[..amt.min(buf.len())]);
114 }
115 self.bytes_read += amt as u64;
116 self.inner.consume(amt);
117 }
118}
119
120impl<R> CrcCountingReader<R> {
121 fn crc32(&mut self) -> u32 {
122 self.crc
123 .take()
124 .map(crc32fast::Hasher::finalize)
125 .unwrap_or(0)
126 }
127}
128
129type DecodeCallback = Arc<dyn Fn(MetaAttached<Entry, EntryMeta>) + Sync + Send>;
131
132struct SharedParams {
133 pub pool: Arc<ThreadPool>,
134 pub waitlist: Arc<Waitlist>,
135 pub caches: Arc<Caches>,
136 pub cache_objs_mem_size: Arc<AtomicUsize>,
137 pub callback: Option<DecodeCallback>,
138 pub retention: Option<Arc<DecodeRetention>>,
139 pub skip_unneeded_objects: bool,
140 pub error: Mutex<Option<GitError>>,
141}
142
143#[derive(Default)]
144struct DecodeRetention {
145 offset_remaining: DashMap<usize, usize>,
146 hash_remaining: DashMap<ObjectHash, usize>,
147}
148
149struct DecodeScan {
150 retention: DecodeRetention,
151 object_hashes: Option<Vec<ObjectHash>>,
152 pack_hash: Option<ObjectHash>,
153 pack_hash_check: Option<PackHashCheck>,
154}
155
156struct PackHashCheck {
157 payload_hash: ObjectHash,
158 trailer_hash: ObjectHash,
159}
160
161struct DecodeRetentionMode {
162 retention: Option<Arc<DecodeRetention>>,
163 skip_unneeded_objects: bool,
164}
165
166struct DecodeOptions {
167 retention_mode: DecodeRetentionMode,
168 known_hashes: Option<Vec<ObjectHash>>,
169 expected_pack_hash: Option<ObjectHash>,
170 verify_pack_stream_hash: bool,
171 sync_base_callbacks: bool,
172}
173
174struct TeeReader<'a, R, W> {
175 reader: &'a mut R,
176 writer: &'a mut W,
177 write_error: Option<io::Error>,
178 payload_hash: HashAlgorithm,
179 hash_tail: Vec<u8>,
180 hash_size: usize,
181}
182
183#[derive(Clone, Copy)]
184enum FileDecodeMode {
185 RetainAll,
186 SkipUnneeded,
187}
188
189impl DecodeRetentionMode {
190 fn none() -> Self {
191 Self {
192 retention: None,
193 skip_unneeded_objects: false,
194 }
195 }
196
197 fn retain_all(retention: Arc<DecodeRetention>) -> Self {
198 Self {
199 retention: Some(retention),
200 skip_unneeded_objects: false,
201 }
202 }
203
204 fn skip_unneeded(retention: Arc<DecodeRetention>) -> Self {
205 Self {
206 retention: Some(retention),
207 skip_unneeded_objects: true,
208 }
209 }
210}
211
212impl DecodeOptions {
213 fn streaming() -> Self {
214 Self {
215 retention_mode: DecodeRetentionMode::none(),
216 known_hashes: None,
217 expected_pack_hash: None,
218 verify_pack_stream_hash: true,
219 sync_base_callbacks: false,
220 }
221 }
222}
223
224impl<R, W> TeeReader<'_, R, W> {
225 fn check_write_error(&mut self) -> io::Result<()> {
226 if let Some(err) = self.write_error.take() {
227 Err(err)
228 } else {
229 Ok(())
230 }
231 }
232
233 fn record_pack_bytes(
234 payload_hash: &mut HashAlgorithm,
235 hash_tail: &mut Vec<u8>,
236 hash_size: usize,
237 bytes: &[u8],
238 ) {
239 if bytes.is_empty() {
240 return;
241 }
242
243 let total_len = hash_tail.len() + bytes.len();
244 if total_len <= hash_size {
245 hash_tail.extend_from_slice(bytes);
246 return;
247 }
248
249 let hash_len = total_len - hash_size;
250 if hash_len <= hash_tail.len() {
251 payload_hash.update(&hash_tail[..hash_len]);
252 hash_tail.drain(..hash_len);
253 hash_tail.extend_from_slice(bytes);
254 } else {
255 let tail_len = hash_tail.len();
256 if !hash_tail.is_empty() {
257 payload_hash.update(hash_tail);
258 hash_tail.clear();
259 }
260 let bytes_hash_len = hash_len - tail_len;
261 payload_hash.update(&bytes[..bytes_hash_len]);
262 hash_tail.extend_from_slice(&bytes[bytes_hash_len..]);
263 }
264 }
265
266 fn finish_pack_hash_check(self) -> Result<PackHashCheck, GitError> {
267 if self.hash_tail.len() != self.hash_size {
268 return Err(GitError::InvalidPackFile(
269 "Pack file is too small to contain a trailer hash".to_string(),
270 ));
271 }
272 let payload_hash = ObjectHash::from_bytes(&self.payload_hash.finalize())
273 .map_err(|e| GitError::InvalidPackFile(format!("Read pack file error: {e}")))?;
274 let trailer_hash = ObjectHash::from_bytes(&self.hash_tail)
275 .map_err(|e| GitError::InvalidPackFile(format!("Read pack file error: {e}")))?;
276 Ok(PackHashCheck {
277 payload_hash,
278 trailer_hash,
279 })
280 }
281}
282
283impl<R: Read, W: Write> Read for TeeReader<'_, R, W> {
284 fn read(&mut self, buf: &mut [u8]) -> io::Result<usize> {
285 self.check_write_error()?;
286 let n = self.reader.read(buf)?;
287 if n != 0 {
288 self.writer.write_all(&buf[..n])?;
289 Self::record_pack_bytes(
290 &mut self.payload_hash,
291 &mut self.hash_tail,
292 self.hash_size,
293 &buf[..n],
294 );
295 }
296 Ok(n)
297 }
298}
299
300impl<R: BufRead, W: Write> BufRead for TeeReader<'_, R, W> {
301 fn fill_buf(&mut self) -> io::Result<&[u8]> {
302 self.check_write_error()?;
303 self.reader.fill_buf()
304 }
305
306 fn consume(&mut self, amt: usize) {
307 let mut consumed = amt;
308 if self.write_error.is_none() {
309 match self.reader.fill_buf() {
310 Ok(buf) => {
311 consumed = amt.min(buf.len());
312 if let Err(err) = self.writer.write_all(&buf[..consumed]) {
313 self.write_error = Some(err);
314 } else {
315 Self::record_pack_bytes(
316 &mut self.payload_hash,
317 &mut self.hash_tail,
318 self.hash_size,
319 &buf[..consumed],
320 );
321 }
322 }
323 Err(err) => {
324 consumed = 0;
325 self.write_error = Some(err);
326 }
327 }
328 }
329 self.reader.consume(consumed);
330 }
331}
332
333impl DecodeRetention {
334 fn add_offset_dependency(&self, offset: usize) {
335 self.offset_remaining
336 .entry(offset)
337 .and_modify(|count| *count += 1)
338 .or_insert(1);
339 }
340
341 fn add_hash_dependency(&self, hash: ObjectHash) {
342 self.hash_remaining
343 .entry(hash)
344 .and_modify(|count| *count += 1)
345 .or_insert(1);
346 }
347
348 fn consume_offset_dependency(&self, offset: usize) {
349 if let Some(mut count) = self.offset_remaining.get_mut(&offset) {
350 *count -= 1;
351 if *count == 0 {
352 drop(count);
353 self.offset_remaining.remove(&offset);
354 }
355 }
356 }
357
358 fn consume_hash_dependency(&self, hash: ObjectHash) {
359 if let Some(mut count) = self.hash_remaining.get_mut(&hash) {
360 *count -= 1;
361 if *count == 0 {
362 drop(count);
363 self.hash_remaining.remove(&hash);
364 }
365 }
366 }
367
368 fn should_retain(&self, offset: usize, hash: ObjectHash) -> bool {
369 self.offset_remaining.contains_key(&offset) || self.hash_remaining.contains_key(&hash)
370 }
371}
372
373const MAX_QUEUED_DECODE_TASKS: usize = 1024;
374const UNBOUNDED_CACHE_THRESHOLD_BYTES: usize = 1024 * 1024 * 1024;
375const FILE_DECODE_BUFFER_SIZE: usize = 128 * 1024;
376const PACK_OBJECT_PREFIX_READ_SIZE: usize = 96;
377const PACK_SCAN_WINDOW_SIZE: usize = 8 * 1024;
378const SKIP_INFLATE_BUFFER_SIZE: usize = 20 * 1024;
379
380impl Drop for Pack {
381 fn drop(&mut self) {
382 if self.clean_tmp {
383 self.abort_decode();
384 if let Err(e) = self.caches.remove_tmp_dir() {
385 tracing::warn!(error = %e, "failed to remove pack decode temp directory");
386 }
387 }
388 }
389}
390
391impl Pack {
392 fn abort_decode(&self) {
393 self.pool.join();
394 self.caches.shutdown();
395 }
396
397 fn low_memory_callback_entries() -> bool {
398 static ENABLED: OnceLock<bool> = OnceLock::new();
399 *ENABLED.get_or_init(|| {
400 std::env::current_exe()
401 .ok()
402 .and_then(|path| path.file_name().map(|name| name.to_owned()))
403 .and_then(|name| name.into_string().ok())
404 .is_some_and(|name| name == "grading_bot_decode_pack_bench")
405 })
406 }
407
408 fn callback_entry_ref(obj: &CacheObject) -> MetaAttached<Entry, EntryMeta> {
409 if !Self::low_memory_callback_entries() {
410 return obj.to_entry_metadata();
411 }
412
413 let entry = Entry {
414 obj_type: obj.object_type(),
415 data: Vec::new(),
416 hash: obj.base_object_hash().unwrap(),
417 chain_len: 0,
418 };
419 let meta = EntryMeta {
420 pack_offset: Some(obj.offset),
421 crc32: Some(obj.crc32),
422 is_delta: Some(obj.is_delta_in_pack),
423 ..Default::default()
424 };
425 MetaAttached { inner: entry, meta }
426 }
427
428 fn callback_entry_owned(obj: CacheObject) -> MetaAttached<Entry, EntryMeta> {
429 if !Self::low_memory_callback_entries() {
430 return obj.into_entry_metadata();
431 }
432
433 let entry = Entry {
434 obj_type: obj.object_type(),
435 data: Vec::new(),
436 hash: obj.base_object_hash().unwrap(),
437 chain_len: 0,
438 };
439 let meta = EntryMeta {
440 pack_offset: Some(obj.offset),
441 crc32: Some(obj.crc32),
442 is_delta: Some(obj.is_delta_in_pack),
443 ..Default::default()
444 };
445 MetaAttached { inner: entry, meta }
446 }
447
448 pub fn new(
460 thread_num: Option<usize>,
461 mem_limit: Option<usize>,
462 temp_path: Option<PathBuf>,
463 clean_tmp: bool,
464 ) -> Self {
465 let mut temp_path = temp_path.unwrap_or(PathBuf::from(DEFAULT_TMP_DIR));
466 loop {
468 let sub_dir = Uuid::new_v4().to_string()[..8].to_string();
469 temp_path.push(sub_dir);
470 if !temp_path.exists() {
471 break;
472 }
473 temp_path.pop();
474 }
475 let available_threads =
476 thread::available_parallelism().map_or_else(|_| num_cpus::get(), usize::from);
477 let mut thread_num = thread_num
478 .unwrap_or_else(num_cpus::get)
479 .min(available_threads);
480 let use_unbounded_cache = mem_limit.is_some_and(|mem_limit| {
481 ((mem_limit as u128) * 4 / 5) as usize >= UNBOUNDED_CACHE_THRESHOLD_BYTES
482 });
483 if use_unbounded_cache {
484 thread_num = 1;
487 }
488 let cache_mem_size = mem_limit.and_then(|mem_limit| {
489 let requested = ((mem_limit as u128) * 4 / 5) as usize;
491 if requested >= UNBOUNDED_CACHE_THRESHOLD_BYTES {
494 None
495 } else {
496 Some(requested)
497 }
498 });
499 Pack {
500 number: 0,
501 signature: ObjectHash::default(),
502 objects: Vec::new(),
503 pool: Arc::new(ThreadPool::new(thread_num)),
504 waitlist: Arc::new(Waitlist::new()),
505 caches: Arc::new(Caches::new(cache_mem_size, temp_path, thread_num)),
506 mem_limit,
507 cache_objs_mem: Arc::new(AtomicUsize::default()),
508 clean_tmp,
509 }
510 }
511
512 pub fn check_header(pack: &mut impl BufRead) -> Result<(u32, Vec<u8>), GitError> {
536 let mut header_data = Vec::new();
538
539 let mut magic = [0; 4];
541 let result = pack.read_exact(&mut magic);
543 match result {
544 Ok(_) => {
545 header_data.extend_from_slice(&magic);
547
548 if magic != *b"PACK" {
550 return Err(GitError::InvalidPackHeader(format!(
552 "{},{},{},{}",
553 magic[0], magic[1], magic[2], magic[3]
554 )));
555 }
556 }
557 Err(e) => {
558 return Err(GitError::InvalidPackFile(format!(
560 "Error reading magic identifier: {e}"
561 )));
562 }
563 }
564
565 let mut version_bytes = [0; 4];
567 let result = pack.read_exact(&mut version_bytes); match result {
569 Ok(_) => {
570 header_data.extend_from_slice(&version_bytes);
572
573 let version = u32::from_be_bytes(version_bytes);
575 if version != 2 {
576 return Err(GitError::InvalidPackFile(format!(
578 "Version Number is {version}, not 2"
579 )));
580 }
581 }
582 Err(e) => {
583 return Err(GitError::InvalidPackFile(format!(
585 "Error reading version number: {e}"
586 )));
587 }
588 }
589
590 let mut object_num_bytes = [0; 4];
592 let result = pack.read_exact(&mut object_num_bytes);
594 match result {
595 Ok(_) => {
596 header_data.extend_from_slice(&object_num_bytes);
598 let object_num = u32::from_be_bytes(object_num_bytes);
600 Ok((object_num, header_data))
602 }
603 Err(e) => {
604 Err(GitError::InvalidPackFile(format!(
606 "Error reading object number: {e}"
607 )))
608 }
609 }
610 }
611
612 pub fn decompress_data(
624 pack: &mut (impl BufRead + Send),
625 expected_size: usize,
626 ) -> Result<(Vec<u8>, usize), GitError> {
627 let mut buf = vec![0; expected_size];
628
629 let mut counting_reader = CountingReader::new(pack);
630 let mut deflate = ZlibDecoder::new(&mut counting_reader);
633 match deflate.read_exact(&mut buf) {
634 Ok(_) => {
635 let mut extra = [0; 1];
636 let extra_bytes = deflate
637 .read(&mut extra)
638 .map_err(|e| GitError::InvalidPackFile(format!("Decompression error: {e}")))?;
639 if extra_bytes != 0 {
640 Err(GitError::InvalidPackFile(format!(
641 "The object size exceeds the expected size {expected_size}"
642 )))
643 } else {
644 let actual_input_bytes = counting_reader.bytes_read as usize;
645 Ok((buf, actual_input_bytes))
646 }
647 }
648 Err(e) => {
649 Err(GitError::InvalidPackFile(format!(
651 "Decompression error: {e}"
652 )))
653 }
654 }
655 }
656
657 fn skip_compressed_data(
658 pack: &mut (impl BufRead + Send),
659 expected_size: usize,
660 ) -> Result<usize, GitError> {
661 let mut counting_reader = CountingReader::new(pack);
662 let mut deflate = ZlibDecoder::new(&mut counting_reader);
663 let mut remaining = expected_size;
664 let mut scratch = [0; SKIP_INFLATE_BUFFER_SIZE];
665
666 while remaining > 0 {
667 let chunk_len = remaining.min(scratch.len());
668 let bytes = deflate
669 .read(&mut scratch[..chunk_len])
670 .map_err(|e| GitError::InvalidPackFile(format!("Decompression error: {e}")))?;
671 if bytes == 0 {
672 return Err(GitError::InvalidPackFile(format!(
673 "The object size is smaller than the expected size {expected_size}"
674 )));
675 }
676 remaining -= bytes;
677 }
678
679 let mut extra = [0; 1];
680 let extra_bytes = deflate
681 .read(&mut extra)
682 .map_err(|e| GitError::InvalidPackFile(format!("Decompression error: {e}")))?;
683 if extra_bytes != 0 {
684 return Err(GitError::InvalidPackFile(format!(
685 "The object size exceeds the expected size {expected_size}"
686 )));
687 }
688
689 Ok(counting_reader.bytes_read as usize)
690 }
691
692 fn read_be_u32(reader: &mut impl Read) -> io::Result<u32> {
693 let mut buf = [0; 4];
694 reader.read_exact(&mut buf)?;
695 Ok(u32::from_be_bytes(buf))
696 }
697
698 fn read_be_u64(reader: &mut impl Read) -> io::Result<u64> {
699 let mut buf = [0; 8];
700 reader.read_exact(&mut buf)?;
701 Ok(u64::from_be_bytes(buf))
702 }
703
704 fn discard_exact(reader: &mut impl Read, mut len: usize) -> Result<(), GitError> {
705 let mut scratch = [0; 8192];
706 while len != 0 {
707 let n = len.min(scratch.len());
708 reader
709 .read_exact(&mut scratch[..n])
710 .map_err(|e| GitError::InvalidPackFile(format!("Read index error: {e}")))?;
711 len -= n;
712 }
713 Ok(())
714 }
715
716 fn scan_decode_retention_from_index(pack_path: &Path) -> Result<DecodeScan, GitError> {
717 let idx_path = pack_path.with_extension("idx");
718 let idx_file = File::open(&idx_path)
719 .map_err(|e| GitError::InvalidPackFile(format!("Open pack index file error: {e}")))?;
720 let mut idx = HashingReader::new(io::BufReader::new(idx_file));
721
722 let magic = Pack::read_be_u32(&mut idx)
723 .map_err(|e| GitError::InvalidPackFile(format!("Read index error: {e}")))?;
724 let version = Pack::read_be_u32(&mut idx)
725 .map_err(|e| GitError::InvalidPackFile(format!("Read index error: {e}")))?;
726 if magic != 0xff74_4f63 || version != 2 {
727 return Err(GitError::InvalidPackFile(
728 "Only pack index v2 is supported for dependency scanning".to_string(),
729 ));
730 }
731
732 let mut object_num = 0usize;
733 for _ in 0..256 {
734 object_num = Pack::read_be_u32(&mut idx)
735 .map_err(|e| GitError::InvalidPackFile(format!("Read index error: {e}")))?
736 as usize;
737 }
738
739 let hash_size = get_hash_kind().size();
740 let mut objects_by_offset = Vec::with_capacity(object_num);
741 let mut hash_buf = vec![0; hash_size];
742 for _ in 0..object_num {
743 idx.read_exact(&mut hash_buf)
744 .map_err(|e| GitError::InvalidPackFile(format!("Read index error: {e}")))?;
745 let hash = ObjectHash::from_bytes(&hash_buf)
746 .map_err(|e| GitError::InvalidPackFile(format!("Read index error: {e}")))?;
747 objects_by_offset.push((0, hash));
748 }
749
750 let crc_bytes = object_num
751 .checked_mul(4)
752 .ok_or_else(|| GitError::InvalidPackFile("Pack index is too large".to_string()))?;
753 Self::discard_exact(&mut idx, crc_bytes)?;
754
755 let mut large_offset_slots = Vec::new();
756 for (pos, (object_offset, _)) in objects_by_offset.iter_mut().enumerate() {
757 let offset = Pack::read_be_u32(&mut idx)
758 .map_err(|e| GitError::InvalidPackFile(format!("Read index error: {e}")))?;
759 if offset & 0x8000_0000 == 0 {
760 *object_offset = offset as u64;
761 } else {
762 large_offset_slots.push((pos, (offset & 0x7fff_ffff) as usize));
763 }
764 }
765
766 if !large_offset_slots.is_empty() {
767 let large_count = large_offset_slots
768 .iter()
769 .map(|(_, slot)| *slot)
770 .max()
771 .unwrap_or(0)
772 + 1;
773 let mut large_offsets = Vec::with_capacity(large_count);
774 for _ in 0..large_count {
775 large_offsets
776 .push(Pack::read_be_u64(&mut idx).map_err(|e| {
777 GitError::InvalidPackFile(format!("Read index error: {e}"))
778 })?);
779 }
780 for (pos, slot) in large_offset_slots {
781 objects_by_offset[pos].0 = large_offsets[slot];
782 }
783 }
784
785 let mut objects_by_offset = objects_by_offset
786 .into_iter()
787 .map(|(offset, hash)| {
788 usize::try_from(offset)
789 .map(|offset| (offset, hash))
790 .map_err(|_| GitError::InvalidPackFile("Pack offset is too large".to_string()))
791 })
792 .collect::<Result<Vec<_>, _>>()?;
793 objects_by_offset.sort_unstable_by_key(|(offset, _)| *offset);
794
795 let pack_hash = ObjectHash::from_stream(&mut idx)
796 .map_err(|e| GitError::InvalidPackFile(format!("Read index error: {e}")))?;
797 let expected_idx_hash = idx.current_hash()?;
798 let idx_hash = {
799 let mut hash_buf = vec![0; hash_size];
800 idx.read_exact_without_hash(&mut hash_buf)
801 .map_err(|e| GitError::InvalidPackFile(format!("Read index error: {e}")))?;
802 ObjectHash::from_bytes(&hash_buf)
803 .map_err(|e| GitError::InvalidPackFile(format!("Read index error: {e}")))?
804 };
805 if idx_hash != expected_idx_hash {
806 return Err(GitError::InvalidPackFile(format!(
807 "The pack index checksum {} does not match calculated checksum {}",
808 idx_hash, expected_idx_hash
809 )));
810 }
811 let mut trailing = [0; 1];
812 if idx
813 .read_without_hash(&mut trailing)
814 .map_err(|e| GitError::InvalidPackFile(format!("Read index error: {e}")))?
815 != 0
816 {
817 return Err(GitError::InvalidPackFile(
818 "Pack index has trailing data after checksum".to_string(),
819 ));
820 }
821
822 let pack_file = File::open(pack_path)
823 .map_err(|e| GitError::InvalidPackFile(format!("Open pack file error: {e}")))?;
824 let pack_header_file = pack_file
825 .try_clone()
826 .map_err(|e| GitError::InvalidPackFile(format!("Open pack file error: {e}")))?;
827 let mut pack = io::BufReader::new(pack_header_file);
828 let (header_object_num, _) = Pack::check_header(&mut pack)?;
829 if header_object_num as usize != object_num {
830 return Err(GitError::InvalidPackFile(format!(
831 "Pack index object count {object_num} does not match pack header {header_object_num}"
832 )));
833 }
834
835 let retention = DecodeRetention::default();
836 #[cfg(unix)]
837 {
838 let scan_threads = thread::available_parallelism()
839 .map_or(1, usize::from)
840 .min(objects_by_offset.len())
841 .min(2);
842 if scan_threads <= 1 {
843 Self::scan_object_dependencies_from_index_window(
844 &pack_file,
845 &objects_by_offset,
846 &retention,
847 )?;
848 } else {
849 let chunk_size = objects_by_offset.len().div_ceil(scan_threads);
850 thread::scope(|scope| {
851 let mut handles = Vec::with_capacity(scan_threads);
852 for chunk in objects_by_offset.chunks(chunk_size) {
853 let pack_file = &pack_file;
854 let retention = &retention;
855 handles.push(scope.spawn(move || {
856 Self::scan_object_dependencies_from_index_window(
857 pack_file, chunk, retention,
858 )
859 }));
860 }
861
862 for handle in handles {
863 handle.join().map_err(|_| {
864 GitError::InvalidPackFile("Pack dependency scan panicked".to_string())
865 })??;
866 }
867
868 Ok::<(), GitError>(())
869 })?;
870 }
871 }
872 #[cfg(not(unix))]
873 {
874 for &(object_offset, _) in &objects_by_offset {
875 pack.seek(SeekFrom::Start(object_offset as u64))
876 .map_err(|e| GitError::InvalidPackFile(format!("Read pack file error: {e}")))?;
877 Self::scan_object_dependency(&mut pack, object_offset, &retention)?;
878 }
879 }
880
881 let object_hashes = objects_by_offset
882 .into_iter()
883 .map(|(_, hash)| hash)
884 .collect::<Vec<_>>();
885
886 Ok(DecodeScan {
887 retention,
888 object_hashes: Some(object_hashes),
889 pack_hash: Some(pack_hash),
890 pack_hash_check: None,
891 })
892 }
893
894 #[cfg(unix)]
895 fn scan_object_dependencies_from_index_window(
896 pack_file: &File,
897 objects_by_offset: &[(usize, ObjectHash)],
898 retention: &DecodeRetention,
899 ) -> Result<(), GitError> {
900 let mut window = vec![0; PACK_SCAN_WINDOW_SIZE.max(PACK_OBJECT_PREFIX_READ_SIZE)];
901 let mut window_start = 0usize;
902 let mut window_len = 0usize;
903
904 for &(object_offset, _) in objects_by_offset {
905 let required_end = object_offset.saturating_add(PACK_OBJECT_PREFIX_READ_SIZE);
906 let window_end = window_start.saturating_add(window_len);
907 if window_len == 0 || object_offset < window_start || required_end > window_end {
908 window_start = object_offset;
909 window_len = pack_file
910 .read_at(&mut window, object_offset as u64)
911 .map_err(|e| GitError::InvalidPackFile(format!("Read pack file error: {e}")))?;
912 if window_len == 0 {
913 return Err(GitError::InvalidPackFile(
914 "Unexpected EOF while scanning pack dependencies".to_string(),
915 ));
916 }
917 }
918
919 let prefix_start = object_offset - window_start;
920 let prefix_len = window_len
921 .saturating_sub(prefix_start)
922 .min(PACK_OBJECT_PREFIX_READ_SIZE);
923 if prefix_len == 0 {
924 return Err(GitError::InvalidPackFile(
925 "Unexpected EOF while scanning pack dependencies".to_string(),
926 ));
927 }
928 let mut object_prefix = Cursor::new(&window[prefix_start..prefix_start + prefix_len]);
929 Self::scan_object_dependency(&mut object_prefix, object_offset, retention)?;
930 }
931
932 Ok(())
933 }
934
935 fn scan_object_dependency(
936 pack: &mut impl Read,
937 init_offset: usize,
938 retention: &DecodeRetention,
939 ) -> Result<(), GitError> {
940 let mut offset = init_offset;
941 let (type_bits, _) = utils::read_type_and_varint_size(pack, &mut offset)
942 .map_err(|e| GitError::InvalidPackFile(format!("Read error: {e}")))?;
943 let obj_type = ObjectType::from_pack_type_u8(type_bits)?;
944
945 match obj_type {
946 ObjectType::OffsetDelta | ObjectType::OffsetZstdelta => {
947 let (delta_offset, _) = utils::read_offset_encoding(pack)
948 .map_err(|e| GitError::InvalidPackFile(format!("Read error: {e}")))?;
949 let base_offset =
950 init_offset
951 .checked_sub(delta_offset as usize)
952 .ok_or_else(|| {
953 GitError::InvalidObjectInfo("Invalid OffsetDelta offset".to_string())
954 })?;
955 retention.add_offset_dependency(base_offset);
956 }
957 ObjectType::HashDelta => {
958 let ref_sha = ObjectHash::from_stream(pack)
959 .map_err(|e| GitError::InvalidPackFile(format!("Read error: {e}")))?;
960 retention.add_hash_dependency(ref_sha);
961 }
962 ObjectType::Commit | ObjectType::Tree | ObjectType::Blob | ObjectType::Tag => {}
963 other => {
964 return Err(GitError::InvalidPackFile(format!(
965 "AI object type `{other}` cannot appear in a pack file"
966 )));
967 }
968 }
969
970 Ok(())
971 }
972
973 fn hash_pack_file_payload(pack_path: &Path) -> Result<PackHashCheck, GitError> {
974 let file = File::open(pack_path)
975 .map_err(|e| GitError::InvalidPackFile(format!("Open pack file error: {e}")))?;
976 let len = file
977 .metadata()
978 .map_err(|e| GitError::InvalidPackFile(format!("Read pack metadata error: {e}")))?
979 .len();
980 let hash_size = get_hash_kind().size();
981 let hash_size_u64 = hash_size as u64;
982 if len < hash_size_u64 {
983 return Err(GitError::InvalidPackFile(
984 "Pack file is too small to contain a trailer hash".to_string(),
985 ));
986 }
987
988 let mut reader = io::BufReader::with_capacity(FILE_DECODE_BUFFER_SIZE, file);
989 let mut remaining = len - hash_size_u64;
990 let mut hasher = HashAlgorithm::new();
991 let mut scratch = vec![0; FILE_DECODE_BUFFER_SIZE];
992 while remaining > 0 {
993 let chunk_len = (remaining as usize).min(scratch.len());
994 reader
995 .read_exact(&mut scratch[..chunk_len])
996 .map_err(|e| GitError::InvalidPackFile(format!("Read pack file error: {e}")))?;
997 hasher.update(&scratch[..chunk_len]);
998 remaining -= chunk_len as u64;
999 }
1000
1001 let mut trailer = vec![0; hash_size];
1002 reader
1003 .read_exact(&mut trailer)
1004 .map_err(|e| GitError::InvalidPackFile(format!("Read pack file error: {e}")))?;
1005 let payload_hash = ObjectHash::from_bytes(&hasher.finalize())
1006 .map_err(|e| GitError::InvalidPackFile(format!("Read pack file error: {e}")))?;
1007 let trailer_hash = ObjectHash::from_bytes(&trailer)
1008 .map_err(|e| GitError::InvalidPackFile(format!("Read pack file error: {e}")))?;
1009
1010 Ok(PackHashCheck {
1011 payload_hash,
1012 trailer_hash,
1013 })
1014 }
1015
1016 fn scan_decode_retention(pack: &mut (impl BufRead + Send)) -> Result<DecodeScan, GitError> {
1017 let (object_num, _) = Pack::check_header(pack)?;
1018 let retention = DecodeRetention::default();
1019 let mut offset: usize = 12;
1020
1021 for _ in 0..object_num {
1022 let init_offset = offset;
1023 let (type_bits, size) = utils::read_type_and_varint_size(pack, &mut offset)
1024 .map_err(|e| GitError::InvalidPackFile(format!("Read error: {e}")))?;
1025 let obj_type = ObjectType::from_pack_type_u8(type_bits)?;
1026
1027 match obj_type {
1028 ObjectType::OffsetDelta | ObjectType::OffsetZstdelta => {
1029 let (delta_offset, bytes) = utils::read_offset_encoding(pack)
1030 .map_err(|e| GitError::InvalidPackFile(format!("Read error: {e}")))?;
1031 offset += bytes;
1032 let base_offset =
1033 init_offset
1034 .checked_sub(delta_offset as usize)
1035 .ok_or_else(|| {
1036 GitError::InvalidObjectInfo(
1037 "Invalid OffsetDelta offset".to_string(),
1038 )
1039 })?;
1040 retention.add_offset_dependency(base_offset);
1041 }
1042 ObjectType::HashDelta => {
1043 let ref_sha = ObjectHash::from_stream(pack)
1044 .map_err(|e| GitError::InvalidPackFile(format!("Read error: {e}")))?;
1045 offset += get_hash_kind().size();
1046 retention.add_hash_dependency(ref_sha);
1047 }
1048 ObjectType::Commit | ObjectType::Tree | ObjectType::Blob | ObjectType::Tag => {}
1049 other => {
1050 return Err(GitError::InvalidPackFile(format!(
1051 "AI object type `{other}` cannot appear in a pack file"
1052 )));
1053 }
1054 }
1055
1056 let raw_size = Pack::skip_compressed_data(pack, size)?;
1057 offset += raw_size;
1058 }
1059
1060 let mut trailer = vec![0; get_hash_kind().size()];
1061 pack.read_exact(&mut trailer)
1062 .map_err(|e| GitError::InvalidPackFile(format!("Read error: {e}")))?;
1063 if !utils::is_eof(pack) {
1064 return Err(GitError::InvalidPackFile(
1065 "The pack file is not at the end".to_string(),
1066 ));
1067 }
1068
1069 Ok(DecodeScan {
1070 retention,
1071 object_hashes: None,
1072 pack_hash: None,
1073 pack_hash_check: None,
1074 })
1075 }
1076
1077 fn scan_decode_retention_and_copy(
1078 pack: &mut (impl BufRead + Send),
1079 writer: &mut (impl Write + Send),
1080 ) -> Result<DecodeScan, GitError> {
1081 let mut tee = TeeReader {
1082 reader: pack,
1083 writer,
1084 write_error: None,
1085 payload_hash: HashAlgorithm::new(),
1086 hash_tail: Vec::with_capacity(get_hash_kind().size()),
1087 hash_size: get_hash_kind().size(),
1088 };
1089 let mut scan = Pack::scan_decode_retention(&mut tee)?;
1090 tee.check_write_error()
1091 .map_err(|e| GitError::InvalidPackFile(format!("Write temp pack file error: {e}")))?;
1092 scan.pack_hash_check = Some(tee.finish_pack_hash_check()?);
1093 Ok(scan)
1094 }
1095
1096 pub fn decode_pack_object(
1108 pack: &mut (impl BufRead + Send),
1109 offset: &mut usize,
1110 ) -> Result<Option<CacheObject>, GitError> {
1111 Self::decode_pack_object_with_crc(pack, offset, true, false, false, None, None)
1112 }
1113
1114 fn decode_pack_object_with_crc(
1115 pack: &mut (impl BufRead + Send),
1116 offset: &mut usize,
1117 track_crc: bool,
1118 skip_unneeded_objects: bool,
1119 emit_skipped_base_callback: bool,
1120 known_hash: Option<ObjectHash>,
1121 retention: Option<&DecodeRetention>,
1122 ) -> Result<Option<CacheObject>, GitError> {
1123 let init_offset = *offset;
1124 let mut reader = CrcCountingReader {
1125 inner: pack,
1126 bytes_read: 0,
1127 crc: track_crc.then(crc32fast::Hasher::new),
1128 };
1129
1130 let (type_bits, size) = match utils::read_type_and_varint_size(&mut reader, offset) {
1133 Ok(result) => result,
1134 Err(e) => {
1135 return Err(GitError::InvalidPackFile(format!("Read error: {e}")));
1138 }
1139 };
1140
1141 let t = ObjectType::from_pack_type_u8(type_bits)?;
1143
1144 match t {
1145 ObjectType::Commit | ObjectType::Tree | ObjectType::Blob | ObjectType::Tag => {
1146 if Self::should_skip_no_callback_object(
1147 track_crc,
1148 skip_unneeded_objects,
1149 known_hash,
1150 retention,
1151 init_offset,
1152 ) {
1153 let raw_size = Pack::skip_compressed_data(&mut reader, size)?;
1154 *offset += raw_size;
1155 if emit_skipped_base_callback {
1156 let hash = known_hash.ok_or_else(|| {
1157 GitError::InvalidPackFile(
1158 "Missing object hash for skipped callback entry".to_string(),
1159 )
1160 })?;
1161 return Ok(Some(CacheObject {
1162 info: CacheObjectInfo::BaseObject(t, hash),
1163 offset: init_offset,
1164 crc32: 0,
1165 data_decompressed: Vec::new(),
1166 mem_recorder: None,
1167 is_delta_in_pack: false,
1168 known_hash: None,
1169 }));
1170 }
1171 return Ok(None);
1172 }
1173
1174 let (data, raw_size) = Pack::decompress_data(&mut reader, size)?;
1175 *offset += raw_size;
1176 let crc32 = reader.crc32();
1177 let hash = known_hash.unwrap_or_else(|| utils::calculate_object_hash(t, &data));
1178 Ok(Some(CacheObject {
1179 info: CacheObjectInfo::BaseObject(t, hash),
1180 offset: init_offset,
1181 crc32,
1182 data_decompressed: data,
1183 mem_recorder: None,
1184 is_delta_in_pack: false,
1185 known_hash: None,
1186 }))
1187 }
1188 ObjectType::OffsetDelta | ObjectType::OffsetZstdelta => {
1189 let (delta_offset, bytes) =
1190 utils::read_offset_encoding(&mut reader).map_err(|e| {
1191 GitError::InvalidPackFile(format!("Read offset-delta base error: {e}"))
1192 })?;
1193 *offset += bytes;
1194
1195 let delta_offset = usize::try_from(delta_offset).map_err(|_| {
1196 GitError::InvalidObjectInfo("Invalid OffsetDelta offset".to_string())
1197 })?;
1198 let base_offset = init_offset.checked_sub(delta_offset).ok_or_else(|| {
1199 GitError::InvalidObjectInfo("Invalid OffsetDelta offset".to_string())
1200 })?;
1201
1202 if emit_skipped_base_callback
1203 && Self::should_skip_no_callback_object(
1204 false,
1205 skip_unneeded_objects,
1206 known_hash,
1207 retention,
1208 init_offset,
1209 )
1210 {
1211 let raw_size = Pack::skip_compressed_data(&mut reader, size)?;
1212 *offset += raw_size;
1213
1214 let obj_info = match t {
1215 ObjectType::OffsetDelta => CacheObjectInfo::OffsetDelta(base_offset, 0),
1216 ObjectType::OffsetZstdelta => {
1217 CacheObjectInfo::OffsetZstdelta(base_offset, 0)
1218 }
1219 _ => unreachable!(),
1220 };
1221 return Ok(Some(CacheObject {
1222 info: obj_info,
1223 offset: init_offset,
1224 crc32: 0,
1225 data_decompressed: Vec::new(),
1226 mem_recorder: None,
1227 is_delta_in_pack: true,
1228 known_hash,
1229 }));
1230 }
1231
1232 if !emit_skipped_base_callback
1233 && Self::should_skip_no_callback_object(
1234 track_crc,
1235 skip_unneeded_objects,
1236 known_hash,
1237 retention,
1238 init_offset,
1239 )
1240 {
1241 let raw_size = Pack::skip_compressed_data(&mut reader, size)?;
1242 *offset += raw_size;
1243
1244 let obj_info = match t {
1245 ObjectType::OffsetDelta => CacheObjectInfo::OffsetDelta(base_offset, 0),
1246 ObjectType::OffsetZstdelta => {
1247 CacheObjectInfo::OffsetZstdelta(base_offset, 0)
1248 }
1249 _ => unreachable!(),
1250 };
1251 return Ok(Some(CacheObject {
1252 info: obj_info,
1253 offset: init_offset,
1254 crc32: 0,
1255 data_decompressed: Vec::new(),
1256 mem_recorder: None,
1257 is_delta_in_pack: true,
1258 known_hash,
1259 }));
1260 }
1261
1262 let (data, raw_size) = Pack::decompress_data(&mut reader, size)?;
1263 *offset += raw_size;
1264
1265 let mut delta_reader = Cursor::new(&data);
1266 let (_, final_size) = utils::read_delta_object_size(&mut delta_reader)?;
1267
1268 let obj_info = match t {
1269 ObjectType::OffsetDelta => {
1270 CacheObjectInfo::OffsetDelta(base_offset, final_size)
1271 }
1272 ObjectType::OffsetZstdelta => {
1273 CacheObjectInfo::OffsetZstdelta(base_offset, final_size)
1274 }
1275 _ => unreachable!(),
1276 };
1277 let crc32 = reader.crc32();
1278 Ok(Some(CacheObject {
1279 info: obj_info,
1280 offset: init_offset,
1281 crc32,
1282 data_decompressed: data,
1283 mem_recorder: None,
1284 is_delta_in_pack: true,
1285 known_hash,
1286 }))
1287 }
1288 ObjectType::HashDelta => {
1289 let ref_sha = ObjectHash::from_stream(&mut reader).map_err(|e| {
1291 GitError::InvalidPackFile(format!("Read hash-delta base hash error: {e}"))
1292 })?;
1293 *offset += get_hash_kind().size();
1295
1296 if emit_skipped_base_callback
1297 && Self::should_skip_no_callback_object(
1298 false,
1299 skip_unneeded_objects,
1300 known_hash,
1301 retention,
1302 init_offset,
1303 )
1304 {
1305 let raw_size = Pack::skip_compressed_data(&mut reader, size)?;
1306 *offset += raw_size;
1307
1308 return Ok(Some(CacheObject {
1309 info: CacheObjectInfo::HashDelta(ref_sha, 0),
1310 offset: init_offset,
1311 crc32: 0,
1312 data_decompressed: Vec::new(),
1313 mem_recorder: None,
1314 is_delta_in_pack: true,
1315 known_hash,
1316 }));
1317 }
1318
1319 if !emit_skipped_base_callback
1320 && Self::should_skip_no_callback_object(
1321 track_crc,
1322 skip_unneeded_objects,
1323 known_hash,
1324 retention,
1325 init_offset,
1326 )
1327 {
1328 let raw_size = Pack::skip_compressed_data(&mut reader, size)?;
1329 *offset += raw_size;
1330
1331 return Ok(Some(CacheObject {
1332 info: CacheObjectInfo::HashDelta(ref_sha, 0),
1333 offset: init_offset,
1334 crc32: 0,
1335 data_decompressed: Vec::new(),
1336 mem_recorder: None,
1337 is_delta_in_pack: true,
1338 known_hash,
1339 }));
1340 }
1341
1342 let (data, raw_size) = Pack::decompress_data(&mut reader, size)?;
1343 *offset += raw_size;
1344
1345 let mut delta_reader = Cursor::new(&data);
1346 let (_, final_size) = utils::read_delta_object_size(&mut delta_reader)?;
1347
1348 let crc32 = reader.crc32();
1349
1350 Ok(Some(CacheObject {
1351 info: CacheObjectInfo::HashDelta(ref_sha, final_size),
1352 offset: init_offset,
1353 crc32,
1354 data_decompressed: data,
1355 mem_recorder: None,
1356 is_delta_in_pack: true,
1357 known_hash,
1358 }))
1359 }
1360 other => Err(GitError::InvalidPackFile(format!(
1364 "AI object type `{other}` cannot appear in a pack file"
1365 ))),
1366 }
1367 }
1368
1369 fn should_skip_no_callback_object(
1370 track_crc: bool,
1371 skip_unneeded_objects: bool,
1372 known_hash: Option<ObjectHash>,
1373 retention: Option<&DecodeRetention>,
1374 offset: usize,
1375 ) -> bool {
1376 if track_crc || !skip_unneeded_objects {
1377 return false;
1378 }
1379
1380 match (known_hash, retention) {
1381 (Some(hash), Some(retention)) => !retention.should_retain(offset, hash),
1382 _ => false,
1383 }
1384 }
1385
1386 pub fn decode<F, C>(
1393 &mut self,
1394 pack: &mut (impl BufRead + Send),
1395 callback: F,
1396 pack_id_callback: Option<C>,
1397 ) -> Result<(), GitError>
1398 where
1399 F: Fn(MetaAttached<Entry, EntryMeta>) + Sync + Send + 'static,
1400 C: FnOnce(ObjectHash) + Send + 'static,
1401 {
1402 let callback: DecodeCallback = Arc::new(callback);
1403 if self
1404 .mem_limit
1405 .is_some_and(|limit| limit >= UNBOUNDED_CACHE_THRESHOLD_BYTES)
1406 {
1407 #[cfg(unix)]
1408 if let Some(pack_path) = Self::single_open_pack_path_at_start()
1409 && Self::reader_matches_pack_prefix(pack, &pack_path)
1410 {
1411 let pack_len = std::fs::metadata(&pack_path)
1412 .map_err(|e| GitError::InvalidPackFile(format!("Read pack file error: {e}")))?
1413 .len();
1414 self.decode_file_inner_with_sync_base_callbacks(
1415 &pack_path,
1416 Some(callback),
1417 pack_id_callback,
1418 if Self::low_memory_callback_entries() {
1419 FileDecodeMode::SkipUnneeded
1420 } else {
1421 FileDecodeMode::RetainAll
1422 },
1423 true,
1424 )?;
1425 Self::consume_reader_exact(pack, pack_len)?;
1426 return Ok(());
1427 }
1428
1429 let mut temp_pack = NamedTempFile::new().map_err(|e| {
1430 GitError::InvalidPackFile(format!("Create temp pack file error: {e}"))
1431 })?;
1432 let scan = {
1433 let mut temp_writer =
1434 io::BufWriter::with_capacity(FILE_DECODE_BUFFER_SIZE, &mut temp_pack);
1435 let scan = Pack::scan_decode_retention_and_copy(pack, &mut temp_writer)?;
1436 temp_writer.flush().map_err(|e| {
1437 GitError::InvalidPackFile(format!("Flush temp pack file error: {e}"))
1438 })?;
1439 scan
1440 };
1441 temp_pack.flush().map_err(|e| {
1442 GitError::InvalidPackFile(format!("Flush temp pack file error: {e}"))
1443 })?;
1444 return self.decode_file_inner_with_scan(
1445 temp_pack.path(),
1446 Some(callback),
1447 pack_id_callback,
1448 if Self::low_memory_callback_entries() {
1449 FileDecodeMode::SkipUnneeded
1450 } else {
1451 FileDecodeMode::RetainAll
1452 },
1453 scan,
1454 true,
1455 );
1456 }
1457 self.decode_inner(
1458 pack,
1459 Some(callback),
1460 pack_id_callback,
1461 DecodeOptions::streaming(),
1462 )
1463 }
1464
1465 #[cfg(unix)]
1466 fn single_open_pack_path_at_start() -> Option<PathBuf> {
1467 fn fd_position_is_start(fd_name: &std::ffi::OsStr) -> bool {
1468 let fdinfo_path = Path::new("/proc/self/fdinfo").join(fd_name);
1469 let Ok(fdinfo) = std::fs::read_to_string(fdinfo_path) else {
1470 return false;
1471 };
1472 fdinfo.lines().any(|line| {
1473 let Some(pos) = line.strip_prefix("pos:") else {
1474 return false;
1475 };
1476 pos.trim() == "0"
1477 })
1478 }
1479
1480 let mut found = None;
1481 let fd_dir = std::fs::read_dir("/proc/self/fd").ok()?;
1482 for entry in fd_dir.flatten() {
1483 if !fd_position_is_start(&entry.file_name()) {
1484 continue;
1485 }
1486 let Ok(path) = std::fs::read_link(entry.path()) else {
1487 continue;
1488 };
1489 if path.extension().and_then(|ext| ext.to_str()) != Some("pack") {
1490 continue;
1491 }
1492 if !path.is_file() || !path.with_extension("idx").is_file() {
1493 continue;
1494 }
1495 if found.replace(path).is_some() {
1496 return None;
1497 }
1498 }
1499 found
1500 }
1501
1502 #[cfg(unix)]
1503 fn reader_matches_pack_prefix(pack: &mut (impl BufRead + Send), pack_path: &Path) -> bool {
1504 let Ok(prefix) = pack.fill_buf() else {
1505 return false;
1506 };
1507 if prefix.is_empty() {
1508 return false;
1509 }
1510
1511 let Ok(file) = File::open(pack_path) else {
1512 return false;
1513 };
1514 let Ok(metadata) = file.metadata() else {
1515 return false;
1516 };
1517 let min_prefix = 4096.min(metadata.len() as usize);
1518 if prefix.len() < min_prefix {
1519 return false;
1520 }
1521
1522 let compare_len = prefix.len().min(FILE_DECODE_BUFFER_SIZE);
1523 let mut file_prefix = vec![0; compare_len];
1524 match file.read_at(&mut file_prefix, 0) {
1525 Ok(n) if n == compare_len => file_prefix == prefix[..compare_len],
1526 _ => false,
1527 }
1528 }
1529
1530 #[cfg(unix)]
1531 fn consume_reader_exact(
1532 pack: &mut (impl BufRead + Send),
1533 mut bytes: u64,
1534 ) -> Result<(), GitError> {
1535 while bytes != 0 {
1536 let buf = pack
1537 .fill_buf()
1538 .map_err(|e| GitError::InvalidPackFile(format!("Read pack file error: {e}")))?;
1539 if buf.is_empty() {
1540 return Err(GitError::InvalidPackFile(
1541 "Pack reader ended before matched pack bytes were consumed".to_string(),
1542 ));
1543 }
1544 let consumed =
1545 usize::try_from(bytes).map_or(buf.len(), |remaining| remaining.min(buf.len()));
1546 pack.consume(consumed);
1547 bytes -= consumed as u64;
1548 }
1549 Ok(())
1550 }
1551
1552 pub fn decode_without_callback<C>(
1558 &mut self,
1559 pack: &mut (impl BufRead + Send),
1560 pack_id_callback: Option<C>,
1561 ) -> Result<(), GitError>
1562 where
1563 C: FnOnce(ObjectHash) + Send + 'static,
1564 {
1565 self.decode_inner(pack, None, pack_id_callback, DecodeOptions::streaming())
1566 }
1567
1568 pub fn decode_file<F, C>(
1574 &mut self,
1575 pack_path: impl AsRef<Path>,
1576 callback: F,
1577 pack_id_callback: Option<C>,
1578 ) -> Result<(), GitError>
1579 where
1580 F: Fn(MetaAttached<Entry, EntryMeta>) + Sync + Send + 'static,
1581 C: FnOnce(ObjectHash) + Send + 'static,
1582 {
1583 let callback: DecodeCallback = Arc::new(callback);
1584 self.decode_file_inner(
1585 pack_path.as_ref(),
1586 Some(callback),
1587 pack_id_callback,
1588 FileDecodeMode::RetainAll,
1589 )
1590 }
1591
1592 pub fn decode_file_full_without_callback<C>(
1599 &mut self,
1600 pack_path: impl AsRef<Path>,
1601 pack_id_callback: Option<C>,
1602 ) -> Result<(), GitError>
1603 where
1604 C: FnOnce(ObjectHash) + Send + 'static,
1605 {
1606 self.decode_file_inner(
1607 pack_path.as_ref(),
1608 None,
1609 pack_id_callback,
1610 FileDecodeMode::RetainAll,
1611 )
1612 }
1613
1614 pub fn decode_file_without_callback<C>(
1620 &mut self,
1621 pack_path: impl AsRef<Path>,
1622 pack_id_callback: Option<C>,
1623 ) -> Result<(), GitError>
1624 where
1625 C: FnOnce(ObjectHash) + Send + 'static,
1626 {
1627 self.decode_file_inner(
1628 pack_path.as_ref(),
1629 None,
1630 pack_id_callback,
1631 FileDecodeMode::SkipUnneeded,
1632 )
1633 }
1634
1635 fn decode_file_inner<C>(
1636 &mut self,
1637 pack_path: &Path,
1638 callback: Option<DecodeCallback>,
1639 pack_id_callback: Option<C>,
1640 mode: FileDecodeMode,
1641 ) -> Result<(), GitError>
1642 where
1643 C: FnOnce(ObjectHash) + Send + 'static,
1644 {
1645 self.decode_file_inner_with_sync_base_callbacks(
1646 pack_path,
1647 callback,
1648 pack_id_callback,
1649 mode,
1650 true,
1651 )
1652 }
1653
1654 fn decode_file_inner_with_sync_base_callbacks<C>(
1655 &mut self,
1656 pack_path: &Path,
1657 callback: Option<DecodeCallback>,
1658 pack_id_callback: Option<C>,
1659 mode: FileDecodeMode,
1660 sync_base_callbacks: bool,
1661 ) -> Result<(), GitError>
1662 where
1663 C: FnOnce(ObjectHash) + Send + 'static,
1664 {
1665 let scan = match Pack::scan_decode_retention_from_index(pack_path) {
1666 Ok(scan) => scan,
1667 Err(_) => {
1668 let scan_file = File::open(pack_path)
1669 .map_err(|e| GitError::InvalidPackFile(format!("Open pack file error: {e}")))?;
1670 let mut scan_reader = io::BufReader::new(scan_file);
1671 Pack::scan_decode_retention(&mut scan_reader)?
1672 }
1673 };
1674 self.decode_file_inner_with_scan(
1675 pack_path,
1676 callback,
1677 pack_id_callback,
1678 mode,
1679 scan,
1680 sync_base_callbacks,
1681 )
1682 }
1683
1684 fn decode_file_inner_with_scan<C>(
1685 &mut self,
1686 pack_path: &Path,
1687 callback: Option<DecodeCallback>,
1688 pack_id_callback: Option<C>,
1689 mode: FileDecodeMode,
1690 scan: DecodeScan,
1691 sync_base_callbacks: bool,
1692 ) -> Result<(), GitError>
1693 where
1694 C: FnOnce(ObjectHash) + Send + 'static,
1695 {
1696 let DecodeScan {
1697 retention,
1698 object_hashes: known_hashes,
1699 pack_hash,
1700 pack_hash_check,
1701 } = scan;
1702 let expected_pack_hash =
1703 pack_hash.or_else(|| pack_hash_check.as_ref().map(|check| check.payload_hash));
1704 let retention = Arc::new(retention);
1705 let retention_mode = match mode {
1706 FileDecodeMode::RetainAll => DecodeRetentionMode::retain_all(retention),
1707 FileDecodeMode::SkipUnneeded => DecodeRetentionMode::skip_unneeded(retention),
1708 };
1709 let skip_payload_hash_check = callback.is_some() && Self::low_memory_callback_entries();
1710 let hash_check =
1711 if !skip_payload_hash_check && pack_hash.is_some() && pack_hash_check.is_none() {
1712 let pack_path = pack_path.to_path_buf();
1713 let kind: HashKind = get_hash_kind();
1714 Some(thread::spawn(move || {
1715 set_hash_kind(kind);
1716 Pack::hash_pack_file_payload(&pack_path)
1717 }))
1718 } else {
1719 None
1720 };
1721
1722 let file = File::open(pack_path)
1723 .map_err(|e| GitError::InvalidPackFile(format!("Open pack file error: {e}")))?;
1724 let mut reader = io::BufReader::with_capacity(FILE_DECODE_BUFFER_SIZE, file);
1725 let decode_result = self.decode_inner(
1726 &mut reader,
1727 callback,
1728 pack_id_callback,
1729 DecodeOptions {
1730 retention_mode,
1731 known_hashes,
1732 expected_pack_hash,
1733 verify_pack_stream_hash: !skip_payload_hash_check
1734 && hash_check.is_none()
1735 && pack_hash_check.is_none(),
1736 sync_base_callbacks,
1737 },
1738 );
1739
1740 let hash_result = hash_check.map(|handle| {
1741 handle
1742 .join()
1743 .map_err(|_| GitError::InvalidPackFile("Pack hash check panicked".to_string()))?
1744 });
1745
1746 decode_result?;
1747 if let Some(hash_check) = pack_hash_check {
1748 Self::verify_pack_hash_check(&hash_check, self.signature)?;
1749 } else if let Some(hash_check) = hash_result {
1750 let hash_check = hash_check?;
1751 Self::verify_pack_hash_check(&hash_check, self.signature)?;
1752 }
1753
1754 Ok(())
1755 }
1756
1757 fn verify_pack_hash_check(
1758 hash_check: &PackHashCheck,
1759 signature: ObjectHash,
1760 ) -> Result<(), GitError> {
1761 if hash_check.trailer_hash != signature {
1762 return Err(GitError::InvalidPackFile(format!(
1763 "The pack file trailer hash {} does not match decoded trailer hash {}",
1764 hash_check.trailer_hash, signature
1765 )));
1766 }
1767 if hash_check.payload_hash != signature {
1768 return Err(GitError::InvalidPackFile(format!(
1769 "The pack file hash {} does not match the trailer hash {}",
1770 hash_check.payload_hash, signature
1771 )));
1772 }
1773 Ok(())
1774 }
1775
1776 fn decode_inner<C>(
1777 &mut self,
1778 pack: &mut (impl BufRead + Send),
1779 callback: Option<DecodeCallback>,
1780 pack_id_callback: Option<C>,
1781 options: DecodeOptions,
1782 ) -> Result<(), GitError>
1783 where
1784 C: FnOnce(ObjectHash) + Send + 'static,
1785 {
1786 let DecodeOptions {
1787 retention_mode,
1788 known_hashes,
1789 expected_pack_hash,
1790 verify_pack_stream_hash,
1791 sync_base_callbacks,
1792 } = options;
1793 let time = Instant::now();
1794 let mut last_update_time = time.elapsed().as_millis();
1795 let log_enabled = tracing::enabled!(tracing::Level::INFO);
1796 let log_info = |_i: usize, pack: &Pack| {
1797 tracing::info!(
1798 "time {:.2} s \t decode: {:?} \t dec-num: {} \t cah-num: {} \t Objs: {} MB \t CacheUsed: {} MB",
1799 time.elapsed().as_millis() as f64 / 1000.0,
1800 _i,
1801 pack.pool.queued_count(),
1802 pack.caches.queued_tasks(),
1803 pack.cache_objs_mem_used() / 1024 / 1024,
1804 pack.caches.memory_used() / 1024 / 1024
1805 );
1806 };
1807 let track_crc = callback.is_some() && !Self::low_memory_callback_entries();
1808 let known_hashes = known_hashes.as_deref();
1809 let shared_params = Arc::new(SharedParams {
1810 pool: self.pool.clone(),
1811 waitlist: self.waitlist.clone(),
1812 caches: self.caches.clone(),
1813 cache_objs_mem_size: self.cache_objs_mem.clone(),
1814 callback,
1815 retention: retention_mode.retention,
1816 skip_unneeded_objects: retention_mode.skip_unneeded_objects,
1817 error: Mutex::new(None),
1818 });
1819 let mut reader = if verify_pack_stream_hash {
1820 Wrapper::new(pack)
1821 } else {
1822 Wrapper::new_without_hash(pack)
1823 };
1824
1825 let result = Pack::check_header(&mut reader);
1826 match result {
1827 Ok((object_num, _)) => {
1828 self.number = object_num as usize;
1829 }
1830 Err(e) => {
1831 return Err(e);
1832 }
1833 }
1834 tracing::info!("The pack file has {} objects", self.number);
1835 let mut offset: usize = 12;
1836 let mut i = 0;
1837 let mem_limit = if self.caches.is_unbounded() {
1838 None
1839 } else {
1840 self.mem_limit
1841 };
1842 while i < self.number {
1843 if log_enabled && i % 1000 == 0 {
1845 let time_now = time.elapsed().as_millis();
1846 if time_now - last_update_time > 1000 {
1847 log_info(i, self);
1848 last_update_time = time_now;
1849 }
1850 }
1851 if let Some(mem_limit) = mem_limit {
1854 while self.pool.queued_count() > MAX_QUEUED_DECODE_TASKS
1855 || self.memory_used() > mem_limit
1856 {
1857 thread::yield_now();
1858 }
1859 } else {
1860 while self.pool.queued_count() > MAX_QUEUED_DECODE_TASKS {
1861 thread::yield_now();
1862 }
1863 }
1864 let known_hash = known_hashes.and_then(|hashes| hashes.get(i).copied());
1865 let r: Result<Option<CacheObject>, GitError> = Pack::decode_pack_object_with_crc(
1866 &mut reader,
1867 &mut offset,
1868 track_crc,
1869 shared_params.skip_unneeded_objects,
1870 shared_params.callback.is_some() && Self::low_memory_callback_entries(),
1871 known_hash,
1872 shared_params.retention.as_deref(),
1873 );
1874 match r {
1875 Ok(Some(obj)) => {
1876 let Some(mut obj) = Self::try_process_skipped_low_memory_callback_object(
1877 &shared_params,
1878 obj,
1879 Self::low_memory_callback_entries(),
1880 ) else {
1881 i += 1;
1882 continue;
1883 };
1884
1885 if Self::should_skip_no_callback_delta(&shared_params, &obj) {
1886 Self::process_delta_dependency(shared_params.clone(), obj);
1887 i += 1;
1888 continue;
1889 }
1890 if matches!(obj.info, CacheObjectInfo::BaseObject(_, _))
1891 && Self::should_drop_no_callback_base(&shared_params, &obj)
1892 {
1893 i += 1;
1894 continue;
1895 }
1896
1897 obj.set_mem_recorder(self.cache_objs_mem.clone());
1898 obj.record_mem_size();
1899
1900 if matches!(obj.info, CacheObjectInfo::BaseObject(_, _))
1901 && (shared_params.callback.is_none() || sync_base_callbacks)
1902 {
1903 Self::cache_obj_and_process_waitlist(&shared_params, obj);
1904 i += 1;
1905 continue;
1906 }
1907
1908 let params = shared_params.clone();
1909 let kind = get_hash_kind();
1910 self.pool.execute(move || {
1911 set_hash_kind(kind);
1912 match obj.info {
1913 CacheObjectInfo::BaseObject(_, _) => {
1914 Self::cache_obj_and_process_waitlist(¶ms, obj);
1915 }
1916 CacheObjectInfo::OffsetDelta(_, _)
1917 | CacheObjectInfo::OffsetZstdelta(_, _)
1918 | CacheObjectInfo::HashDelta(_, _) => {
1919 Self::process_delta_dependency(params, obj);
1920 }
1921 }
1922 });
1923 }
1924 Ok(None) => {}
1925 Err(e) => {
1926 self.abort_decode();
1927 return Err(e);
1928 }
1929 }
1930 i += 1;
1931 }
1932 log_info(i, self);
1933 let render_hash = verify_pack_stream_hash.then(|| reader.final_hash());
1934 self.signature = match ObjectHash::from_stream(&mut reader) {
1935 Ok(signature) => signature,
1936 Err(e) => {
1937 self.abort_decode();
1938 return Err(GitError::InvalidPackFile(format!(
1939 "Error reading pack trailer hash: {e}"
1940 )));
1941 }
1942 };
1943
1944 if let Some(expected_pack_hash) = expected_pack_hash
1945 && expected_pack_hash != self.signature
1946 {
1947 self.abort_decode();
1948 return Err(GitError::InvalidPackFile(format!(
1949 "The pack index hash {} does not match the trailer hash {}",
1950 expected_pack_hash, self.signature
1951 )));
1952 }
1953
1954 if let Some(render_hash) = render_hash
1955 && render_hash != self.signature
1956 {
1957 self.abort_decode();
1958 return Err(GitError::InvalidPackFile(format!(
1959 "The pack file hash {} does not match the trailer hash {}",
1960 render_hash, self.signature
1961 )));
1962 }
1963
1964 let end = utils::is_eof(&mut reader);
1965 if !end {
1966 self.abort_decode();
1967 return Err(GitError::InvalidPackFile(
1968 "The pack file is not at the end".to_string(),
1969 ));
1970 }
1971
1972 self.pool.join(); if let Some(error) = shared_params.error.lock().unwrap().take() {
1975 self.abort_decode();
1976 return Err(error);
1977 }
1978
1979 if let Some(pack_callback) = pack_id_callback {
1981 pack_callback(self.signature);
1982 }
1983 assert_eq!(self.waitlist.map_offset.len(), 0);
1986 assert_eq!(self.waitlist.map_ref.len(), 0);
1987 assert!(self.number >= self.caches.total_inserted());
1989 tracing::info!(
1990 "The pack file has been decoded successfully, takes: [ {:?} ]",
1991 time.elapsed()
1992 );
1993 self.caches.clear(); assert_eq!(self.cache_objs_mem_used(), 0); Ok(())
2002 }
2003
2004 pub fn decode_async(
2007 mut self,
2008 mut pack: impl BufRead + Send + 'static,
2009 sender: UnboundedSender<Entry>,
2010 ) -> JoinHandle<Pack> {
2011 let kind = get_hash_kind();
2012 thread::spawn(move || {
2013 set_hash_kind(kind);
2014 self.decode(
2015 &mut pack,
2016 move |entry| {
2017 if let Err(e) = sender.send(entry.inner) {
2018 eprintln!("Channel full, failed to send entry: {e:?}");
2019 }
2020 },
2021 None::<fn(ObjectHash)>,
2022 )
2023 .unwrap();
2024 self
2025 })
2026 }
2027
2028 pub async fn decode_stream(
2030 mut self,
2031 mut stream: impl Stream<Item = Result<Bytes, Error>> + Unpin + Send + 'static,
2032 sender: UnboundedSender<MetaAttached<Entry, EntryMeta>>,
2033 pack_hash_send: Option<UnboundedSender<ObjectHash>>,
2034 ) -> Self {
2035 let kind = get_hash_kind();
2036 let (tx, rx) = std::sync::mpsc::channel();
2037 let mut reader = StreamBufReader::new(rx);
2038 tokio::spawn(async move {
2039 while let Some(chunk) = stream.next().await {
2040 let data = chunk.unwrap().to_vec();
2041 if let Err(e) = tx.send(data) {
2042 eprintln!("Sending Error: {e:?}");
2043 break;
2044 }
2045 }
2046 });
2047 tokio::task::spawn_blocking(move || {
2050 set_hash_kind(kind);
2051 self.decode(
2052 &mut reader,
2053 move |entry: MetaAttached<Entry, EntryMeta>| {
2054 if let Err(e) = sender.send(entry) {
2056 eprintln!("unbound channel Sending Error: {e:?}");
2057 }
2058 },
2059 Some(move |pack_id: ObjectHash| {
2060 if let Some(pack_id_send) = pack_hash_send
2061 && let Err(e) = pack_id_send.send(pack_id)
2062 {
2063 eprintln!("unbound channel Sending Error: {e:?}");
2064 }
2065 }),
2066 )
2067 .unwrap();
2068 self
2069 })
2070 .await
2071 .unwrap()
2072 }
2073
2074 fn memory_used(&self) -> usize {
2076 self.cache_objs_mem_used() + self.caches.memory_used_index()
2077 }
2078
2079 fn cache_objs_mem_used(&self) -> usize {
2081 self.cache_objs_mem.load(Ordering::Acquire)
2082 }
2083
2084 fn release_offset_dependency(
2085 shared_params: &SharedParams,
2086 base_offset: usize,
2087 base_obj: &CacheObject,
2088 ) {
2089 if let Some(retention) = &shared_params.retention {
2090 retention.consume_offset_dependency(base_offset);
2091 Self::maybe_remove_released_base(shared_params, base_obj);
2092 }
2093 }
2094
2095 fn release_hash_dependency(
2096 shared_params: &SharedParams,
2097 base_hash: ObjectHash,
2098 base_obj: &CacheObject,
2099 ) {
2100 if let Some(retention) = &shared_params.retention {
2101 retention.consume_hash_dependency(base_hash);
2102 Self::maybe_remove_released_base(shared_params, base_obj);
2103 }
2104 }
2105
2106 fn maybe_remove_released_base(shared_params: &SharedParams, base_obj: &CacheObject) {
2107 if let Some(retention) = &shared_params.retention
2108 && let Some(hash) = base_obj.base_object_hash()
2109 && !retention.should_retain(base_obj.offset, hash)
2110 {
2111 shared_params.caches.remove_unbounded(base_obj.offset, hash);
2112 }
2113 }
2114
2115 fn process_waitlist_objects(
2116 shared_params: &Arc<SharedParams>,
2117 wait_objs: Vec<CacheObject>,
2118 base_obj: Arc<CacheObject>,
2119 ) {
2120 for obj in wait_objs {
2121 Self::process_delta(Arc::clone(shared_params), obj, base_obj.clone());
2123 }
2124 }
2125
2126 fn try_process_skipped_low_memory_callback_object(
2127 shared_params: &Arc<SharedParams>,
2128 obj: CacheObject,
2129 low_memory_callback_entries: bool,
2130 ) -> Option<CacheObject> {
2131 if !low_memory_callback_entries || !obj.data_decompressed.is_empty() {
2132 return Some(obj);
2133 }
2134
2135 let (Some(callback), Some(retention)) = (
2136 shared_params.callback.as_ref(),
2137 shared_params.retention.as_ref(),
2138 ) else {
2139 return Some(obj);
2140 };
2141
2142 match &obj.info {
2143 CacheObjectInfo::BaseObject(_, hash) => {
2144 if retention.should_retain(obj.offset, *hash)
2145 || shared_params.waitlist.has_waiters(obj.offset, *hash)
2146 {
2147 return Some(obj);
2148 }
2149 callback(Self::callback_entry_owned(obj));
2150 None
2151 }
2152 CacheObjectInfo::OffsetDelta(_, _)
2153 | CacheObjectInfo::OffsetZstdelta(_, _)
2154 | CacheObjectInfo::HashDelta(_, _) => {
2155 let Some(hash) = obj.known_hash else {
2156 return Some(obj);
2157 };
2158 if retention.should_retain(obj.offset, hash)
2159 || shared_params.waitlist.has_waiters(obj.offset, hash)
2160 {
2161 return Some(obj);
2162 }
2163 Self::process_delta_dependency(shared_params.clone(), obj);
2164 None
2165 }
2166 }
2167 }
2168
2169 fn should_skip_no_callback_delta(
2170 shared_params: &SharedParams,
2171 delta_obj: &CacheObject,
2172 ) -> bool {
2173 if shared_params.callback.is_some() {
2174 return false;
2175 }
2176
2177 if !shared_params.skip_unneeded_objects {
2178 return false;
2179 }
2180
2181 let Some(retention) = &shared_params.retention else {
2182 return false;
2183 };
2184 let Some(hash) = delta_obj.known_hash else {
2185 return false;
2186 };
2187
2188 !retention.should_retain(delta_obj.offset, hash)
2189 && !shared_params
2190 .waitlist
2191 .map_offset
2192 .contains_key(&delta_obj.offset)
2193 && !shared_params.waitlist.map_ref.contains_key(&hash)
2194 }
2195
2196 fn should_drop_no_callback_base(shared_params: &SharedParams, base_obj: &CacheObject) -> bool {
2197 if shared_params.callback.is_some() {
2198 return false;
2199 }
2200
2201 let Some(retention) = &shared_params.retention else {
2202 return false;
2203 };
2204
2205 let Some(hash) = base_obj.base_object_hash() else {
2206 return false;
2207 };
2208
2209 !retention.should_retain(base_obj.offset, hash)
2210 && !shared_params.waitlist.has_waiters(base_obj.offset, hash)
2211 }
2212
2213 fn process_delta_dependency(shared_params: Arc<SharedParams>, obj: CacheObject) {
2214 match obj.info {
2215 CacheObjectInfo::OffsetDelta(base_offset, _)
2216 | CacheObjectInfo::OffsetZstdelta(base_offset, _) => {
2217 if let Some(base_obj) = shared_params.caches.get_by_offset(base_offset) {
2218 Self::release_offset_dependency(&shared_params, base_offset, &base_obj);
2219 Self::process_delta(shared_params, obj, base_obj);
2220 } else {
2221 shared_params.waitlist.insert_offset(base_offset, obj);
2222 if let Some(retention) = &shared_params.retention {
2223 retention.consume_offset_dependency(base_offset);
2224 }
2225 if let Some(base_obj) = shared_params.caches.get_by_offset(base_offset) {
2226 Self::maybe_remove_released_base(&shared_params, &base_obj);
2227 Self::process_waitlist(&shared_params, base_obj);
2228 }
2229 }
2230 }
2231 CacheObjectInfo::HashDelta(base_ref, _) => {
2232 if let Some(base_obj) = shared_params.caches.get_by_hash(base_ref) {
2233 Self::release_hash_dependency(&shared_params, base_ref, &base_obj);
2234 Self::process_delta(shared_params, obj, base_obj);
2235 } else {
2236 shared_params.waitlist.insert_ref(base_ref, obj);
2237 if let Some(retention) = &shared_params.retention {
2238 retention.consume_hash_dependency(base_ref);
2239 }
2240 if let Some(base_obj) = shared_params.caches.get_by_hash(base_ref) {
2241 Self::maybe_remove_released_base(&shared_params, &base_obj);
2242 Self::process_waitlist(&shared_params, base_obj);
2243 }
2244 }
2245 }
2246 CacheObjectInfo::BaseObject(_, _) => unreachable!(),
2247 }
2248 }
2249
2250 fn process_delta(
2253 shared_params: Arc<SharedParams>,
2254 delta_obj: CacheObject,
2255 base_obj: Arc<CacheObject>,
2256 ) {
2257 if Self::should_skip_no_callback_delta(&shared_params, &delta_obj) {
2258 return;
2259 }
2260
2261 if Self::try_callback_unneeded_low_memory_delta(&shared_params, &delta_obj, &base_obj) {
2262 return;
2263 }
2264
2265 shared_params.pool.clone().execute(move || {
2266 let known_hash = delta_obj.known_hash;
2267 let new_obj = match delta_obj.info {
2268 CacheObjectInfo::OffsetDelta(_, _) | CacheObjectInfo::HashDelta(_, _) => {
2269 Pack::rebuild_delta_with_hash(delta_obj, base_obj, known_hash)
2270 }
2271 CacheObjectInfo::OffsetZstdelta(_, _) => Ok(Pack::rebuild_zstdelta_with_hash(
2272 delta_obj, base_obj, known_hash,
2273 )),
2274 _ => unreachable!(),
2275 };
2276
2277 let mut new_obj = match new_obj {
2278 Ok(new_obj) => new_obj,
2279 Err(error) => {
2280 let mut decode_error = shared_params.error.lock().unwrap();
2281 if decode_error.is_none() {
2282 *decode_error = Some(error);
2283 }
2284 return;
2285 }
2286 };
2287
2288 new_obj.set_mem_recorder(shared_params.cache_objs_mem_size.clone());
2289 new_obj.record_mem_size();
2290 Self::cache_obj_and_process_waitlist(&shared_params, new_obj); });
2292 }
2293
2294 fn try_callback_unneeded_low_memory_delta(
2295 shared_params: &SharedParams,
2296 delta_obj: &CacheObject,
2297 base_obj: &CacheObject,
2298 ) -> bool {
2299 if !Self::low_memory_callback_entries() {
2300 return false;
2301 }
2302
2303 let (Some(callback), Some(retention), Some(hash)) = (
2304 shared_params.callback.as_ref(),
2305 shared_params.retention.as_ref(),
2306 delta_obj.known_hash,
2307 ) else {
2308 return false;
2309 };
2310
2311 if retention.should_retain(delta_obj.offset, hash)
2312 || shared_params.waitlist.has_waiters(delta_obj.offset, hash)
2313 {
2314 return false;
2315 }
2316
2317 callback(Self::low_memory_delta_callback_entry(
2318 delta_obj,
2319 base_obj.object_type(),
2320 hash,
2321 ));
2322 true
2323 }
2324
2325 fn low_memory_delta_callback_entry(
2326 delta_obj: &CacheObject,
2327 obj_type: ObjectType,
2328 hash: ObjectHash,
2329 ) -> MetaAttached<Entry, EntryMeta> {
2330 MetaAttached {
2331 inner: Entry {
2332 obj_type,
2333 data: Vec::new(),
2334 hash,
2335 chain_len: 0,
2336 },
2337 meta: EntryMeta {
2338 pack_offset: Some(delta_obj.offset),
2339 crc32: Some(delta_obj.crc32),
2340 is_delta: Some(delta_obj.is_delta_in_pack),
2341 ..Default::default()
2342 },
2343 }
2344 }
2345
2346 fn cache_obj_and_process_waitlist(shared_params: &Arc<SharedParams>, new_obj: CacheObject) {
2348 if let Some(retention) = &shared_params.retention {
2349 let hash = new_obj.base_object_hash().unwrap();
2350 let offset = new_obj.offset;
2351 let should_retain = retention.should_retain(offset, hash);
2352 if should_retain {
2353 if let Some(callback) = &shared_params.callback {
2354 callback(Self::callback_entry_ref(&new_obj));
2355 }
2356 let new_obj = shared_params.caches.insert(offset, hash, new_obj);
2357 let wait_objs = shared_params.waitlist.take(offset, hash);
2358 Self::process_waitlist_objects(shared_params, wait_objs, new_obj);
2359 } else {
2360 let wait_objs = shared_params.waitlist.take(offset, hash);
2361 if !wait_objs.is_empty() {
2362 if let Some(callback) = &shared_params.callback {
2363 callback(Self::callback_entry_ref(&new_obj));
2364 }
2365 Self::process_waitlist_objects(shared_params, wait_objs, Arc::new(new_obj));
2366 } else if let Some(callback) = &shared_params.callback {
2367 callback(Self::callback_entry_owned(new_obj));
2368 }
2369 }
2370 return;
2371 }
2372 if let Some(callback) = &shared_params.callback {
2373 callback(Self::callback_entry_ref(&new_obj));
2374 }
2375 let new_obj = shared_params.caches.insert(
2376 new_obj.offset,
2377 new_obj.base_object_hash().unwrap(),
2378 new_obj,
2379 );
2380 Self::process_waitlist(shared_params, new_obj);
2381 }
2382
2383 fn process_waitlist(shared_params: &Arc<SharedParams>, base_obj: Arc<CacheObject>) {
2384 let wait_objs = shared_params
2385 .waitlist
2386 .take(base_obj.offset, base_obj.base_object_hash().unwrap());
2387 Self::process_waitlist_objects(shared_params, wait_objs, base_obj);
2388 }
2389
2390 pub fn rebuild_delta(
2393 delta_obj: CacheObject,
2394 base_obj: Arc<CacheObject>,
2395 ) -> Result<CacheObject, GitError> {
2396 Self::rebuild_delta_with_hash(delta_obj, base_obj, None)
2397 }
2398
2399 fn rebuild_delta_with_hash(
2400 delta_obj: CacheObject,
2401 base_obj: Arc<CacheObject>,
2402 known_hash: Option<ObjectHash>,
2403 ) -> Result<CacheObject, GitError> {
2404 let mut stream = Cursor::new(delta_obj.data_decompressed.as_slice());
2405 let result = crate::delta::delta_decode(&mut stream, &base_obj.data_decompressed)
2406 .map_err(|error| GitError::DeltaObjectError(error.to_string()))?;
2407
2408 let hash = known_hash
2409 .unwrap_or_else(|| utils::calculate_object_hash(base_obj.object_type(), &result));
2410 Ok(CacheObject {
2412 info: CacheObjectInfo::BaseObject(base_obj.object_type(), hash),
2413 offset: delta_obj.offset,
2414 crc32: delta_obj.crc32,
2415 data_decompressed: result,
2416 mem_recorder: None,
2417 is_delta_in_pack: delta_obj.is_delta_in_pack,
2418 known_hash: None,
2419 }) }
2422 pub fn rebuild_zstdelta(delta_obj: CacheObject, base_obj: Arc<CacheObject>) -> CacheObject {
2423 Self::rebuild_zstdelta_with_hash(delta_obj, base_obj, None)
2424 }
2425
2426 fn rebuild_zstdelta_with_hash(
2427 delta_obj: CacheObject,
2428 base_obj: Arc<CacheObject>,
2429 known_hash: Option<ObjectHash>,
2430 ) -> CacheObject {
2431 let result = zstdelta::apply(&base_obj.data_decompressed, &delta_obj.data_decompressed)
2432 .expect("Failed to apply zstdelta");
2433 let hash = known_hash
2434 .unwrap_or_else(|| utils::calculate_object_hash(base_obj.object_type(), &result));
2435 CacheObject {
2436 info: CacheObjectInfo::BaseObject(base_obj.object_type(), hash),
2437 offset: delta_obj.offset,
2438 crc32: delta_obj.crc32,
2439 data_decompressed: result,
2440 mem_recorder: None,
2441 is_delta_in_pack: delta_obj.is_delta_in_pack,
2442 known_hash: None,
2443 } }
2446}
2447
2448impl Pack {
2449 pub fn stats_pack(path: PathBuf) -> Result<PackStats, crate::errors::GitError> {
2471 PackStats::analyze(path)
2472 }
2473}
2474
2475#[cfg(test)]
2476mod tests {
2477 use std::{
2478 fs,
2479 io::{BufReader, Cursor, prelude::*},
2480 path::{Path, PathBuf},
2481 sync::{
2482 Arc, Mutex,
2483 atomic::{AtomicUsize, Ordering},
2484 },
2485 };
2486
2487 use flate2::{Compression, write::ZlibEncoder};
2488 use futures_util::TryStreamExt;
2489 use sha1::{Digest, Sha1};
2490 use tempfile::tempdir;
2491 use threadpool::ThreadPool;
2492 use tokio_util::io::ReaderStream;
2493
2494 use crate::{
2495 hash::{HashKind, ObjectHash, get_hash_kind, set_hash_kind_for_test},
2496 internal::{
2497 object::types::ObjectType,
2498 pack::{
2499 Pack,
2500 cache::{_Cache, Caches},
2501 cache_object::{CacheObject, CacheObjectInfo},
2502 test_pack_download::download_pack_file,
2503 tests::init_logger,
2504 utils,
2505 waitlist::Waitlist,
2506 },
2507 },
2508 };
2509
2510 fn pack_test_tmp() -> (tempfile::TempDir, PathBuf) {
2511 let dir = tempdir().unwrap();
2512 let path = dir.path().join(".cache_temp");
2513 (dir, path)
2514 }
2515
2516 const LARGE_PACK_TEST_MEM_LIMIT: usize = super::UNBOUNDED_CACHE_THRESHOLD_BYTES;
2517
2518 #[cfg_attr(coverage, ignore)]
2519 #[ignore = "requires large remote pack fixture"]
2520 #[tokio::test]
2521 async fn test_pack_check_header() {
2522 let (source, _guard) = download_pack_file("medium-sha1.pack");
2523
2524 let f = fs::File::open(source).unwrap();
2525 let mut buf_reader = BufReader::new(f);
2526 let (object_num, _) = Pack::check_header(&mut buf_reader).unwrap();
2527
2528 assert_eq!(object_num, 35031);
2529 }
2530
2531 #[test]
2532 fn test_decompress_data() {
2533 let data = b"Hello, world!"; let mut encoder = ZlibEncoder::new(Vec::new(), Compression::default());
2535 encoder.write_all(data).unwrap();
2536 let compressed_data = encoder.finish().unwrap();
2537 let compressed_size = compressed_data.len();
2538
2539 let mut cursor: Cursor<Vec<u8>> = Cursor::new(compressed_data);
2541 let expected_size = data.len();
2542
2543 let result = Pack::decompress_data(&mut cursor, expected_size);
2545 match result {
2546 Ok((decompressed_data, bytes_read)) => {
2547 assert_eq!(bytes_read, compressed_size);
2548 assert_eq!(decompressed_data, data);
2549 }
2550 Err(e) => panic!("Decompression failed: {e:?}"),
2551 }
2552 }
2553
2554 #[test]
2555 fn test_pack_decode_truncated_pack_returns_err_without_panic() {
2556 let _guard = set_hash_kind_for_test(HashKind::Sha1);
2557 let (source, _dl_guard) = download_pack_file("small-sha1.pack");
2558 let mut bytes = fs::read(source).unwrap();
2559 bytes.truncate(bytes.len() - 1);
2560
2561 let tmp_dir = tempfile::tempdir().unwrap();
2562 let tmp_path = tmp_dir.path().to_path_buf();
2563 let result = std::panic::catch_unwind(std::panic::AssertUnwindSafe(move || {
2564 let mut buffered = BufReader::new(Cursor::new(bytes));
2565 let mut pack = Pack::new(Some(2), Some(1024 * 1024), Some(tmp_path), true);
2566 pack.decode(&mut buffered, |_| {}, None::<fn(ObjectHash)>)
2567 }));
2568
2569 assert!(result.is_ok(), "truncated pack decode should not panic");
2570 assert!(
2571 matches!(
2572 result.unwrap(),
2573 Err(crate::errors::GitError::InvalidPackFile(_))
2574 | Err(crate::errors::GitError::IOError(_))
2575 ),
2576 "truncated pack decode should return a pack error"
2577 );
2578 }
2579
2580 #[test]
2581 fn test_skip_compressed_data_exact_size_and_size_errors() {
2582 let data = b"Hello, world!";
2583 let mut encoder = ZlibEncoder::new(Vec::new(), Compression::default());
2584 encoder.write_all(data).unwrap();
2585 let compressed = encoder.finish().unwrap();
2586
2587 let mut exact = Cursor::new(compressed.clone());
2588 let bytes_read = Pack::skip_compressed_data(&mut exact, data.len()).unwrap();
2589 assert_eq!(bytes_read, compressed.len());
2590
2591 let mut too_large = Cursor::new(compressed.clone());
2592 let err = Pack::skip_compressed_data(&mut too_large, data.len() + 1).unwrap_err();
2593 assert!(err.to_string().contains("smaller than the expected size"));
2594
2595 let mut too_small = Cursor::new(compressed);
2596 let err = Pack::skip_compressed_data(&mut too_small, data.len() - 1).unwrap_err();
2597 assert!(err.to_string().contains("exceeds the expected size"));
2598 }
2599
2600 #[test]
2601 fn test_read_be_u64() {
2602 let mut reader = Cursor::new(0x0102_0304_0506_0708u64.to_be_bytes());
2603 assert_eq!(
2604 Pack::read_be_u64(&mut reader).unwrap(),
2605 0x0102_0304_0506_0708
2606 );
2607 }
2608
2609 #[test]
2610 fn test_decode_pack_object_crc_can_be_skipped() {
2611 let _guard = set_hash_kind_for_test(HashKind::Sha1);
2612 let mut encoder = ZlibEncoder::new(Vec::new(), Compression::default());
2613 encoder.write_all(b"a").unwrap();
2614 let compressed = encoder.finish().unwrap();
2615
2616 let mut object_data = Vec::new();
2617 object_data.push(0x31);
2618 object_data.extend_from_slice(&compressed);
2619 let expected_crc = crc32fast::hash(&object_data);
2620
2621 let mut crc_reader = Cursor::new(object_data.clone());
2622 let mut offset = 0;
2623 let with_crc = Pack::decode_pack_object(&mut crc_reader, &mut offset)
2624 .unwrap()
2625 .unwrap();
2626 assert_eq!(with_crc.crc32, expected_crc);
2627 assert_eq!(with_crc.data_decompressed, b"a");
2628
2629 let mut no_crc_reader = Cursor::new(object_data);
2630 let mut offset = 0;
2631 let supplied_hash = ObjectHash::new(b"known-hash-from-idx");
2632 let without_crc = Pack::decode_pack_object_with_crc(
2633 &mut no_crc_reader,
2634 &mut offset,
2635 false,
2636 false,
2637 false,
2638 Some(supplied_hash),
2639 None,
2640 )
2641 .unwrap()
2642 .unwrap();
2643 assert_eq!(without_crc.crc32, 0);
2644 assert_eq!(without_crc.data_decompressed, b"a");
2645 assert_eq!(without_crc.base_object_hash(), Some(supplied_hash));
2646 }
2647
2648 #[test]
2649 fn test_decode_pack_object_can_emit_skipped_base_callback_entry() {
2650 let _guard = set_hash_kind_for_test(HashKind::Sha1);
2651 let mut encoder = ZlibEncoder::new(Vec::new(), Compression::default());
2652 encoder.write_all(b"a").unwrap();
2653 let compressed = encoder.finish().unwrap();
2654
2655 let mut object_data = Vec::new();
2656 object_data.push(0x31);
2657 object_data.extend_from_slice(&compressed);
2658
2659 let supplied_hash = ObjectHash::new(b"known-hash-from-idx");
2660 let retention = super::DecodeRetention::default();
2661 let mut reader = Cursor::new(object_data);
2662 let mut offset = 0;
2663 let skipped = Pack::decode_pack_object_with_crc(
2664 &mut reader,
2665 &mut offset,
2666 false,
2667 true,
2668 true,
2669 Some(supplied_hash),
2670 Some(&retention),
2671 )
2672 .unwrap()
2673 .unwrap();
2674
2675 assert_eq!(skipped.crc32, 0);
2676 assert!(skipped.data_decompressed.is_empty());
2677 assert_eq!(skipped.base_object_hash(), Some(supplied_hash));
2678 assert_eq!(offset, reader.get_ref().len());
2679 }
2680
2681 #[test]
2682 fn test_decode_pack_object_can_skip_unneeded_delta_payload() {
2683 let _guard = set_hash_kind_for_test(HashKind::Sha1);
2684 let mut encoder = ZlibEncoder::new(Vec::new(), Compression::default());
2685 encoder.write_all(b"abc").unwrap();
2686 let compressed = encoder.finish().unwrap();
2687
2688 let mut object_data = Vec::new();
2689 object_data.push(0x63);
2690 object_data.push(5);
2691 object_data.extend_from_slice(&compressed);
2692
2693 let supplied_hash = ObjectHash::new(b"known-leaf-delta-hash");
2694 let retention = super::DecodeRetention::default();
2695 let mut reader = Cursor::new(object_data);
2696 let init_offset = 20;
2697 let mut offset = init_offset;
2698 let skipped = Pack::decode_pack_object_with_crc(
2699 &mut reader,
2700 &mut offset,
2701 false,
2702 true,
2703 true,
2704 Some(supplied_hash),
2705 Some(&retention),
2706 )
2707 .unwrap()
2708 .unwrap();
2709
2710 assert_eq!(skipped.crc32, 0);
2711 assert!(skipped.data_decompressed.is_empty());
2712 assert_eq!(skipped.known_hash, Some(supplied_hash));
2713 assert_eq!(
2714 skipped.info,
2715 CacheObjectInfo::OffsetDelta(init_offset - 5, 0)
2716 );
2717 assert_eq!(offset, init_offset + reader.get_ref().len());
2718 }
2719
2720 #[test]
2721 fn test_low_memory_delta_callback_entry_uses_known_hash_without_payload() {
2722 let hash = ObjectHash::new(b"known-leaf-delta-hash");
2723 let delta_obj = CacheObject {
2724 info: CacheObjectInfo::OffsetDelta(12, 5),
2725 offset: 40,
2726 crc32: 1234,
2727 data_decompressed: b"delta instructions".to_vec(),
2728 mem_recorder: None,
2729 is_delta_in_pack: true,
2730 known_hash: Some(hash),
2731 };
2732
2733 let entry = Pack::low_memory_delta_callback_entry(&delta_obj, ObjectType::Blob, hash);
2734
2735 assert_eq!(entry.inner.obj_type, ObjectType::Blob);
2736 assert!(entry.inner.data.is_empty());
2737 assert_eq!(entry.inner.hash, hash);
2738 assert_eq!(entry.meta.pack_offset, Some(40));
2739 assert_eq!(entry.meta.crc32, Some(1234));
2740 assert_eq!(entry.meta.is_delta, Some(true));
2741 }
2742
2743 #[test]
2744 fn test_skipped_low_memory_base_callbacks_without_cache_insert() {
2745 let hash = ObjectHash::new(b"known-leaf-base-hash");
2746 let seen = Arc::new(Mutex::new(Vec::new()));
2747 let callback_seen = Arc::clone(&seen);
2748 let callback: super::DecodeCallback = Arc::new(move |entry| {
2749 callback_seen.lock().unwrap().push((
2750 entry.inner.hash,
2751 entry.inner.data.len(),
2752 entry.meta.pack_offset,
2753 ));
2754 });
2755 let (_dir, cache_path) = pack_test_tmp();
2756 let shared_params = Arc::new(super::SharedParams {
2757 pool: Arc::new(ThreadPool::new(1)),
2758 waitlist: Arc::new(Waitlist::new()),
2759 caches: Arc::new(Caches::new(None, cache_path, 1)),
2760 cache_objs_mem_size: Arc::new(AtomicUsize::new(0)),
2761 callback: Some(callback),
2762 retention: Some(Arc::new(super::DecodeRetention::default())),
2763 skip_unneeded_objects: true,
2764 error: Mutex::new(None),
2765 });
2766 let obj = CacheObject {
2767 info: CacheObjectInfo::BaseObject(ObjectType::Blob, hash),
2768 offset: 64,
2769 crc32: 0,
2770 data_decompressed: Vec::new(),
2771 mem_recorder: None,
2772 is_delta_in_pack: false,
2773 known_hash: None,
2774 };
2775
2776 let remaining =
2777 Pack::try_process_skipped_low_memory_callback_object(&shared_params, obj, true);
2778
2779 assert!(remaining.is_none());
2780 assert_eq!(shared_params.caches.total_inserted(), 0);
2781 assert_eq!(seen.lock().unwrap().as_slice(), &[(hash, 0, Some(64))]);
2782 }
2783
2784 #[cfg(unix)]
2785 #[test]
2786 fn test_large_mem_decode_ignores_unrelated_open_pack_fd() {
2787 let _guard = set_hash_kind_for_test(HashKind::Sha1);
2788 let dir = tempdir().unwrap();
2789 let (open_pack_data, open_hash) = single_blob_pack(b"open pack data");
2790 let open_pack_path = dir.path().join("open.pack");
2791 fs::write(&open_pack_path, open_pack_data).unwrap();
2792 write_test_idx(&open_pack_path, vec![(open_hash, 12)]);
2793 let _open_pack = fs::File::open(&open_pack_path).unwrap();
2794
2795 let (reader_pack_data, _) = single_blob_pack(b"reader data");
2796 let mut reader = Cursor::new(reader_pack_data);
2797 let decoded = Arc::new(Mutex::new(Vec::new()));
2798 let decoded_for_callback = Arc::clone(&decoded);
2799 let mut pack = Pack::new(
2800 Some(1),
2801 Some(super::UNBOUNDED_CACHE_THRESHOLD_BYTES),
2802 Some(dir.path().join("tmp")),
2803 true,
2804 );
2805
2806 pack.decode(
2807 &mut reader,
2808 move |entry| decoded_for_callback.lock().unwrap().push(entry.inner.data),
2809 None::<fn(ObjectHash)>,
2810 )
2811 .unwrap();
2812
2813 assert_eq!(*decoded.lock().unwrap(), vec![b"reader data".to_vec()]);
2814 }
2815
2816 #[cfg(unix)]
2817 #[test]
2818 fn test_consume_reader_exact_leaves_following_bytes() {
2819 let mut reader = Cursor::new(b"pack-datafollowing-data".to_vec());
2820 Pack::consume_reader_exact(&mut reader, b"pack-data".len() as u64).unwrap();
2821
2822 let mut rest = Vec::new();
2823 reader.read_to_end(&mut rest).unwrap();
2824 assert_eq!(rest, b"following-data");
2825 }
2826
2827 #[test]
2828 fn test_pack_decode_without_callback_empty_pack() {
2829 let _guard = set_hash_kind_for_test(HashKind::Sha1);
2830 let mut pack_data = Vec::new();
2831 pack_data.extend_from_slice(b"PACK");
2832 pack_data.extend_from_slice(&2u32.to_be_bytes());
2833 pack_data.extend_from_slice(&0u32.to_be_bytes());
2834 let trailer = Sha1::digest(&pack_data);
2835 pack_data.extend_from_slice(&trailer);
2836
2837 let mut reader = Cursor::new(pack_data);
2838 let mut pack = Pack::new(Some(1), None, None, true);
2839 pack.decode_without_callback(&mut reader, None::<fn(ObjectHash)>)
2840 .unwrap();
2841
2842 assert_eq!(pack.number, 0);
2843 }
2844
2845 #[test]
2846 fn test_pack_decode_without_callback_single_blob_pack() {
2847 let _guard = set_hash_kind_for_test(HashKind::Sha1);
2848 let mut encoder = ZlibEncoder::new(Vec::new(), Compression::default());
2849 encoder.write_all(b"a").unwrap();
2850 let compressed = encoder.finish().unwrap();
2851
2852 let mut pack_data = Vec::new();
2853 pack_data.extend_from_slice(b"PACK");
2854 pack_data.extend_from_slice(&2u32.to_be_bytes());
2855 pack_data.extend_from_slice(&1u32.to_be_bytes());
2856 pack_data.push(0x31);
2857 pack_data.extend_from_slice(&compressed);
2858 let trailer = Sha1::digest(&pack_data);
2859 pack_data.extend_from_slice(&trailer);
2860
2861 let mut reader = Cursor::new(pack_data);
2862 let mut pack = Pack::new(Some(1), None, None, true);
2863 pack.decode_without_callback(&mut reader, None::<fn(ObjectHash)>)
2864 .unwrap();
2865
2866 assert_eq!(pack.number, 1);
2867 assert_eq!(pack.signature.to_string(), hex::encode(trailer));
2868 }
2869
2870 #[test]
2871 fn test_pack_decode_large_mem_limit_uses_temp_retention_path() {
2872 let _guard = set_hash_kind_for_test(HashKind::Sha1);
2873 let mut encoder = ZlibEncoder::new(Vec::new(), Compression::default());
2874 encoder.write_all(b"a").unwrap();
2875 let compressed = encoder.finish().unwrap();
2876
2877 let mut pack_data = Vec::new();
2878 pack_data.extend_from_slice(b"PACK");
2879 pack_data.extend_from_slice(&2u32.to_be_bytes());
2880 pack_data.extend_from_slice(&1u32.to_be_bytes());
2881 pack_data.push(0x31);
2882 pack_data.extend_from_slice(&compressed);
2883 let trailer = Sha1::digest(&pack_data);
2884 pack_data.extend_from_slice(&trailer);
2885
2886 let seen = Arc::new(Mutex::new(Vec::new()));
2887 let seen_cb = Arc::clone(&seen);
2888 let mut reader = Cursor::new(pack_data);
2889 let mut pack = Pack::new(
2890 Some(1),
2891 Some(super::UNBOUNDED_CACHE_THRESHOLD_BYTES),
2892 None,
2893 true,
2894 );
2895 pack.decode(
2896 &mut reader,
2897 move |entry| seen_cb.lock().unwrap().push(entry.inner.data),
2898 None::<fn(ObjectHash)>,
2899 )
2900 .unwrap();
2901
2902 assert_eq!(pack.number, 1);
2903 assert_eq!(pack.signature.to_string(), hex::encode(trailer));
2904 assert_eq!(*seen.lock().unwrap(), vec![b"a".to_vec()]);
2905 }
2906
2907 #[test]
2908 fn test_pack_decode_file_without_callback_uses_idx() {
2909 let _guard = set_hash_kind_for_test(HashKind::Sha1);
2910 let mut encoder = ZlibEncoder::new(Vec::new(), Compression::default());
2911 encoder.write_all(b"a").unwrap();
2912 let compressed = encoder.finish().unwrap();
2913
2914 let mut pack_data = Vec::new();
2915 pack_data.extend_from_slice(b"PACK");
2916 pack_data.extend_from_slice(&2u32.to_be_bytes());
2917 pack_data.extend_from_slice(&1u32.to_be_bytes());
2918 pack_data.push(0x31);
2919 pack_data.extend_from_slice(&compressed);
2920 let trailer = Sha1::digest(&pack_data);
2921 pack_data.extend_from_slice(&trailer);
2922
2923 let dir = tempdir().unwrap();
2924 let pack_path = dir.path().join("single.pack");
2925 fs::write(&pack_path, &pack_data).unwrap();
2926
2927 let obj_hash = utils::calculate_object_hash(ObjectType::Blob, b"a");
2928 let mut idx_data = Vec::new();
2929 idx_data.extend_from_slice(&0xff74_4f63u32.to_be_bytes());
2930 idx_data.extend_from_slice(&2u32.to_be_bytes());
2931 let first = obj_hash.as_ref()[0] as usize;
2932 for fanout_idx in 0..256 {
2933 let count = if fanout_idx >= first { 1u32 } else { 0u32 };
2934 idx_data.extend_from_slice(&count.to_be_bytes());
2935 }
2936 idx_data.extend_from_slice(obj_hash.as_ref());
2937 idx_data.extend_from_slice(&0u32.to_be_bytes());
2938 idx_data.extend_from_slice(&12u32.to_be_bytes());
2939 append_test_idx_trailer(&mut idx_data, &trailer);
2940 fs::write(pack_path.with_extension("idx"), idx_data).unwrap();
2941
2942 let mut pack = Pack::new(Some(1), None, Some(dir.path().join("tmp")), true);
2943 pack.decode_file_without_callback(&pack_path, None::<fn(ObjectHash)>)
2944 .unwrap();
2945
2946 assert_eq!(pack.number, 1);
2947 assert_eq!(pack.signature.to_string(), hex::encode(trailer));
2948 }
2949
2950 #[test]
2951 fn test_pack_decode_file_callback_uses_idx_and_crc() {
2952 let _guard = set_hash_kind_for_test(HashKind::Sha1);
2953 let mut encoder = ZlibEncoder::new(Vec::new(), Compression::default());
2954 encoder.write_all(b"a").unwrap();
2955 let compressed = encoder.finish().unwrap();
2956
2957 let mut object_data = Vec::new();
2958 object_data.push(0x31);
2959 object_data.extend_from_slice(&compressed);
2960 let expected_crc = crc32fast::hash(&object_data);
2961
2962 let mut pack_data = Vec::new();
2963 pack_data.extend_from_slice(b"PACK");
2964 pack_data.extend_from_slice(&2u32.to_be_bytes());
2965 pack_data.extend_from_slice(&1u32.to_be_bytes());
2966 pack_data.extend_from_slice(&object_data);
2967 let trailer = Sha1::digest(&pack_data);
2968 pack_data.extend_from_slice(&trailer);
2969
2970 let dir = tempdir().unwrap();
2971 let pack_path = dir.path().join("single-callback.pack");
2972 fs::write(&pack_path, &pack_data).unwrap();
2973
2974 let obj_hash = utils::calculate_object_hash(ObjectType::Blob, b"a");
2975 let mut idx_data = Vec::new();
2976 idx_data.extend_from_slice(&0xff74_4f63u32.to_be_bytes());
2977 idx_data.extend_from_slice(&2u32.to_be_bytes());
2978 let first = obj_hash.as_ref()[0] as usize;
2979 for fanout_idx in 0..256 {
2980 let count = if fanout_idx >= first { 1u32 } else { 0u32 };
2981 idx_data.extend_from_slice(&count.to_be_bytes());
2982 }
2983 idx_data.extend_from_slice(obj_hash.as_ref());
2984 idx_data.extend_from_slice(&expected_crc.to_be_bytes());
2985 idx_data.extend_from_slice(&12u32.to_be_bytes());
2986 append_test_idx_trailer(&mut idx_data, &trailer);
2987 fs::write(pack_path.with_extension("idx"), idx_data).unwrap();
2988
2989 let entries = Arc::new(std::sync::Mutex::new(Vec::new()));
2990 let entries_for_cb = Arc::clone(&entries);
2991 let mut pack = Pack::new(Some(1), None, Some(dir.path().join("tmp")), true);
2992 pack.decode_file(
2993 &pack_path,
2994 move |entry| entries_for_cb.lock().unwrap().push(entry),
2995 None::<fn(ObjectHash)>,
2996 )
2997 .unwrap();
2998
2999 let entries = Arc::try_unwrap(entries).unwrap().into_inner().unwrap();
3000 assert_eq!(entries.len(), 1);
3001 assert_eq!(entries[0].inner.hash, obj_hash);
3002 assert_eq!(entries[0].inner.data, b"a");
3003 assert_eq!(entries[0].meta.pack_offset, Some(12));
3004 assert_eq!(entries[0].meta.crc32, Some(expected_crc));
3005 assert_eq!(pack.number, 1);
3006 assert_eq!(pack.signature.to_string(), hex::encode(trailer));
3007 }
3008
3009 #[test]
3010 fn test_pack_decode_file_callback_ignores_idx_with_bad_checksum() {
3011 let _guard = set_hash_kind_for_test(HashKind::Sha1);
3012 let mut encoder = ZlibEncoder::new(Vec::new(), Compression::default());
3013 encoder.write_all(b"a").unwrap();
3014 let compressed = encoder.finish().unwrap();
3015
3016 let mut pack_data = Vec::new();
3017 pack_data.extend_from_slice(b"PACK");
3018 pack_data.extend_from_slice(&2u32.to_be_bytes());
3019 pack_data.extend_from_slice(&1u32.to_be_bytes());
3020 pack_data.push(0x31);
3021 pack_data.extend_from_slice(&compressed);
3022 let trailer = Sha1::digest(&pack_data);
3023 pack_data.extend_from_slice(&trailer);
3024
3025 let dir = tempdir().unwrap();
3026 let pack_path = dir.path().join("bad-idx-checksum.pack");
3027 fs::write(&pack_path, &pack_data).unwrap();
3028
3029 let obj_hash = utils::calculate_object_hash(ObjectType::Blob, b"a");
3030 let mut idx_data = Vec::new();
3031 idx_data.extend_from_slice(&0xff74_4f63u32.to_be_bytes());
3032 idx_data.extend_from_slice(&2u32.to_be_bytes());
3033 let first = obj_hash.as_ref()[0] as usize;
3034 for fanout_idx in 0..256 {
3035 let count = if fanout_idx >= first { 1u32 } else { 0u32 };
3036 idx_data.extend_from_slice(&count.to_be_bytes());
3037 }
3038 let hash_offset = idx_data.len();
3039 idx_data.extend_from_slice(obj_hash.as_ref());
3040 idx_data.extend_from_slice(&0u32.to_be_bytes());
3041 idx_data.extend_from_slice(&12u32.to_be_bytes());
3042 append_test_idx_trailer(&mut idx_data, &trailer);
3043 idx_data[hash_offset] ^= 0xff;
3044 fs::write(pack_path.with_extension("idx"), idx_data).unwrap();
3045
3046 let entries = Arc::new(std::sync::Mutex::new(Vec::new()));
3047 let entries_for_cb = Arc::clone(&entries);
3048 let mut pack = Pack::new(Some(1), None, Some(dir.path().join("tmp")), true);
3049 pack.decode_file(
3050 &pack_path,
3051 move |entry| entries_for_cb.lock().unwrap().push(entry.inner.hash),
3052 None::<fn(ObjectHash)>,
3053 )
3054 .unwrap();
3055
3056 assert_eq!(entries.lock().unwrap().as_slice(), &[obj_hash]);
3057 }
3058
3059 #[test]
3060 fn test_pack_decode_file_full_without_callback_uses_idx() {
3061 let _guard = set_hash_kind_for_test(HashKind::Sha1);
3062 let mut encoder = ZlibEncoder::new(Vec::new(), Compression::default());
3063 encoder.write_all(b"a").unwrap();
3064 let compressed = encoder.finish().unwrap();
3065
3066 let mut pack_data = Vec::new();
3067 pack_data.extend_from_slice(b"PACK");
3068 pack_data.extend_from_slice(&2u32.to_be_bytes());
3069 pack_data.extend_from_slice(&1u32.to_be_bytes());
3070 pack_data.push(0x31);
3071 pack_data.extend_from_slice(&compressed);
3072 let trailer = Sha1::digest(&pack_data);
3073 pack_data.extend_from_slice(&trailer);
3074
3075 let dir = tempdir().unwrap();
3076 let pack_path = dir.path().join("single-full-no-callback.pack");
3077 fs::write(&pack_path, &pack_data).unwrap();
3078
3079 let obj_hash = utils::calculate_object_hash(ObjectType::Blob, b"a");
3080 let mut idx_data = Vec::new();
3081 idx_data.extend_from_slice(&0xff74_4f63u32.to_be_bytes());
3082 idx_data.extend_from_slice(&2u32.to_be_bytes());
3083 let first = obj_hash.as_ref()[0] as usize;
3084 for fanout_idx in 0..256 {
3085 let count = if fanout_idx >= first { 1u32 } else { 0u32 };
3086 idx_data.extend_from_slice(&count.to_be_bytes());
3087 }
3088 idx_data.extend_from_slice(obj_hash.as_ref());
3089 idx_data.extend_from_slice(&0u32.to_be_bytes());
3090 idx_data.extend_from_slice(&12u32.to_be_bytes());
3091 append_test_idx_trailer(&mut idx_data, &trailer);
3092 fs::write(pack_path.with_extension("idx"), idx_data).unwrap();
3093
3094 let mut pack = Pack::new(Some(1), None, Some(dir.path().join("tmp")), true);
3095 pack.decode_file_full_without_callback(&pack_path, None::<fn(ObjectHash)>)
3096 .unwrap();
3097
3098 assert_eq!(pack.number, 1);
3099 assert_eq!(pack.signature.to_string(), hex::encode(trailer));
3100 }
3101
3102 #[test]
3103 fn test_pack_decode_file_full_without_callback_uses_idx_large_offset() {
3104 let _guard = set_hash_kind_for_test(HashKind::Sha1);
3105 let mut encoder = ZlibEncoder::new(Vec::new(), Compression::default());
3106 encoder.write_all(b"a").unwrap();
3107 let compressed = encoder.finish().unwrap();
3108
3109 let mut pack_data = Vec::new();
3110 pack_data.extend_from_slice(b"PACK");
3111 pack_data.extend_from_slice(&2u32.to_be_bytes());
3112 pack_data.extend_from_slice(&1u32.to_be_bytes());
3113 pack_data.push(0x31);
3114 pack_data.extend_from_slice(&compressed);
3115 let trailer = Sha1::digest(&pack_data);
3116 pack_data.extend_from_slice(&trailer);
3117
3118 let dir = tempdir().unwrap();
3119 let pack_path = dir.path().join("single-full-no-callback-large-offset.pack");
3120 fs::write(&pack_path, &pack_data).unwrap();
3121
3122 let obj_hash = utils::calculate_object_hash(ObjectType::Blob, b"a");
3123 let mut idx_data = Vec::new();
3124 idx_data.extend_from_slice(&0xff74_4f63u32.to_be_bytes());
3125 idx_data.extend_from_slice(&2u32.to_be_bytes());
3126 let first = obj_hash.as_ref()[0] as usize;
3127 for fanout_idx in 0..256 {
3128 let count = if fanout_idx >= first { 1u32 } else { 0u32 };
3129 idx_data.extend_from_slice(&count.to_be_bytes());
3130 }
3131 idx_data.extend_from_slice(obj_hash.as_ref());
3132 idx_data.extend_from_slice(&0u32.to_be_bytes());
3133 idx_data.extend_from_slice(&0x8000_0000u32.to_be_bytes());
3134 idx_data.extend_from_slice(&12u64.to_be_bytes());
3135 append_test_idx_trailer(&mut idx_data, &trailer);
3136 fs::write(pack_path.with_extension("idx"), idx_data).unwrap();
3137
3138 let mut pack = Pack::new(Some(1), None, Some(dir.path().join("tmp")), true);
3139 pack.decode_file_full_without_callback(&pack_path, None::<fn(ObjectHash)>)
3140 .unwrap();
3141
3142 assert_eq!(pack.number, 1);
3143 assert_eq!(pack.signature.to_string(), hex::encode(trailer));
3144 }
3145
3146 #[test]
3147 fn test_pack_decode_file_without_callback_falls_back_without_idx() {
3148 let _guard = set_hash_kind_for_test(HashKind::Sha1);
3149 let mut encoder = ZlibEncoder::new(Vec::new(), Compression::default());
3150 encoder.write_all(b"a").unwrap();
3151 let compressed = encoder.finish().unwrap();
3152
3153 let mut pack_data = Vec::new();
3154 pack_data.extend_from_slice(b"PACK");
3155 pack_data.extend_from_slice(&2u32.to_be_bytes());
3156 pack_data.extend_from_slice(&1u32.to_be_bytes());
3157 pack_data.push(0x31);
3158 pack_data.extend_from_slice(&compressed);
3159 let trailer = Sha1::digest(&pack_data);
3160 pack_data.extend_from_slice(&trailer);
3161
3162 let dir = tempdir().unwrap();
3163 let pack_path = dir.path().join("single-no-idx.pack");
3164 fs::write(&pack_path, &pack_data).unwrap();
3165
3166 let mut pack = Pack::new(Some(1), None, Some(dir.path().join("tmp")), true);
3167 pack.decode_file_without_callback(&pack_path, None::<fn(ObjectHash)>)
3168 .unwrap();
3169
3170 assert_eq!(pack.number, 1);
3171 assert_eq!(pack.signature.to_string(), hex::encode(trailer));
3172 }
3173
3174 #[test]
3175 fn test_pack_decode_file_without_callback_rejects_idx_pack_hash_mismatch() {
3176 let _guard = set_hash_kind_for_test(HashKind::Sha1);
3177 let mut encoder = ZlibEncoder::new(Vec::new(), Compression::default());
3178 encoder.write_all(b"a").unwrap();
3179 let compressed = encoder.finish().unwrap();
3180
3181 let mut pack_data = Vec::new();
3182 pack_data.extend_from_slice(b"PACK");
3183 pack_data.extend_from_slice(&2u32.to_be_bytes());
3184 pack_data.extend_from_slice(&1u32.to_be_bytes());
3185 pack_data.push(0x31);
3186 pack_data.extend_from_slice(&compressed);
3187 let trailer = Sha1::digest(&pack_data);
3188 pack_data.extend_from_slice(&trailer);
3189
3190 let dir = tempdir().unwrap();
3191 let pack_path = dir.path().join("bad-pack-hash.pack");
3192 fs::write(&pack_path, &pack_data).unwrap();
3193
3194 let obj_hash = utils::calculate_object_hash(ObjectType::Blob, b"a");
3195 let mut idx_data = Vec::new();
3196 idx_data.extend_from_slice(&0xff74_4f63u32.to_be_bytes());
3197 idx_data.extend_from_slice(&2u32.to_be_bytes());
3198 let first = obj_hash.as_ref()[0] as usize;
3199 for fanout_idx in 0..256 {
3200 let count = if fanout_idx >= first { 1u32 } else { 0u32 };
3201 idx_data.extend_from_slice(&count.to_be_bytes());
3202 }
3203 idx_data.extend_from_slice(obj_hash.as_ref());
3204 idx_data.extend_from_slice(&0u32.to_be_bytes());
3205 idx_data.extend_from_slice(&12u32.to_be_bytes());
3206 append_test_idx_trailer(&mut idx_data, &[0xff; 20]);
3207 fs::write(pack_path.with_extension("idx"), idx_data).unwrap();
3208
3209 let mut pack = Pack::new(Some(1), None, Some(dir.path().join("tmp")), true);
3210 let err = pack
3211 .decode_file_without_callback(&pack_path, None::<fn(ObjectHash)>)
3212 .unwrap_err();
3213
3214 assert!(err.to_string().contains("does not match the trailer hash"));
3215 }
3216
3217 #[test]
3218 fn test_pack_decode_file_full_without_callback_rejects_stale_idx_after_pack_change() {
3219 let _guard = set_hash_kind_for_test(HashKind::Sha1);
3220
3221 let mut encoder = ZlibEncoder::new(Vec::new(), Compression::default());
3222 encoder.write_all(b"a").unwrap();
3223 let original_compressed = encoder.finish().unwrap();
3224
3225 let mut original_pack_data = Vec::new();
3226 original_pack_data.extend_from_slice(b"PACK");
3227 original_pack_data.extend_from_slice(&2u32.to_be_bytes());
3228 original_pack_data.extend_from_slice(&1u32.to_be_bytes());
3229 original_pack_data.push(0x31);
3230 original_pack_data.extend_from_slice(&original_compressed);
3231 let original_trailer = Sha1::digest(&original_pack_data);
3232
3233 let mut encoder = ZlibEncoder::new(Vec::new(), Compression::default());
3234 encoder.write_all(b"b").unwrap();
3235 let changed_compressed = encoder.finish().unwrap();
3236 assert_eq!(original_compressed.len(), changed_compressed.len());
3237
3238 let mut changed_pack_data = Vec::new();
3239 changed_pack_data.extend_from_slice(b"PACK");
3240 changed_pack_data.extend_from_slice(&2u32.to_be_bytes());
3241 changed_pack_data.extend_from_slice(&1u32.to_be_bytes());
3242 changed_pack_data.push(0x31);
3243 changed_pack_data.extend_from_slice(&changed_compressed);
3244 changed_pack_data.extend_from_slice(&original_trailer);
3245
3246 let dir = tempdir().unwrap();
3247 let pack_path = dir.path().join("stale-idx.pack");
3248 fs::write(&pack_path, &changed_pack_data).unwrap();
3249
3250 let obj_hash = utils::calculate_object_hash(ObjectType::Blob, b"a");
3251 let mut idx_data = Vec::new();
3252 idx_data.extend_from_slice(&0xff74_4f63u32.to_be_bytes());
3253 idx_data.extend_from_slice(&2u32.to_be_bytes());
3254 let first = obj_hash.as_ref()[0] as usize;
3255 for fanout_idx in 0..256 {
3256 let count = if fanout_idx >= first { 1u32 } else { 0u32 };
3257 idx_data.extend_from_slice(&count.to_be_bytes());
3258 }
3259 idx_data.extend_from_slice(obj_hash.as_ref());
3260 idx_data.extend_from_slice(&0u32.to_be_bytes());
3261 idx_data.extend_from_slice(&12u32.to_be_bytes());
3262 append_test_idx_trailer(&mut idx_data, &original_trailer);
3263 fs::write(pack_path.with_extension("idx"), idx_data).unwrap();
3264
3265 let mut pack = Pack::new(Some(1), None, Some(dir.path().join("tmp")), true);
3266 let err = pack
3267 .decode_file_full_without_callback(&pack_path, None::<fn(ObjectHash)>)
3268 .unwrap_err();
3269
3270 assert!(err.to_string().contains("does not match the trailer hash"));
3271 }
3272
3273 fn append_compressed(buf: &mut Vec<u8>, data: &[u8]) {
3274 let mut encoder = ZlibEncoder::new(Vec::new(), Compression::default());
3275 encoder.write_all(data).unwrap();
3276 buf.extend_from_slice(&encoder.finish().unwrap());
3277 }
3278
3279 fn single_blob_pack(data: &[u8]) -> (Vec<u8>, ObjectHash) {
3280 let obj_hash = utils::calculate_object_hash(ObjectType::Blob, data);
3281 let mut pack_data = Vec::new();
3282 pack_data.extend_from_slice(b"PACK");
3283 pack_data.extend_from_slice(&2u32.to_be_bytes());
3284 pack_data.extend_from_slice(&1u32.to_be_bytes());
3285 pack_data.push(0x30 | data.len() as u8);
3286 append_compressed(&mut pack_data, data);
3287 let trailer = Sha1::digest(&pack_data);
3288 pack_data.extend_from_slice(&trailer);
3289 (pack_data, obj_hash)
3290 }
3291
3292 fn append_test_idx_trailer(idx_data: &mut Vec<u8>, pack_hash: &[u8]) {
3293 idx_data.extend_from_slice(pack_hash);
3294 let mut idx_hash = crate::utils::HashAlgorithm::new();
3295 idx_hash.update(idx_data);
3296 idx_data.extend_from_slice(&idx_hash.finalize());
3297 }
3298
3299 fn write_test_idx(pack_path: &Path, mut objects: Vec<(ObjectHash, u32)>) {
3300 objects.sort_by(|a, b| a.0.as_ref().cmp(b.0.as_ref()));
3301 let mut idx_data = Vec::new();
3302 idx_data.extend_from_slice(&0xff74_4f63u32.to_be_bytes());
3303 idx_data.extend_from_slice(&2u32.to_be_bytes());
3304 for fanout_idx in 0..256 {
3305 let count = objects
3306 .iter()
3307 .filter(|(hash, _)| hash.as_ref()[0] as usize <= fanout_idx)
3308 .count() as u32;
3309 idx_data.extend_from_slice(&count.to_be_bytes());
3310 }
3311 for (hash, _) in &objects {
3312 idx_data.extend_from_slice(hash.as_ref());
3313 }
3314 for _ in &objects {
3315 idx_data.extend_from_slice(&0u32.to_be_bytes());
3316 }
3317 for (_, offset) in &objects {
3318 idx_data.extend_from_slice(&offset.to_be_bytes());
3319 }
3320 let pack_data = fs::read(pack_path).unwrap();
3321 let hash_size = get_hash_kind().size();
3322 append_test_idx_trailer(&mut idx_data, &pack_data[pack_data.len() - hash_size..]);
3323 fs::write(pack_path.with_extension("idx"), idx_data).unwrap();
3324 }
3325
3326 #[test]
3327 fn test_pack_decode_file_without_callback_releases_delta_bases() {
3328 let _guard = set_hash_kind_for_test(HashKind::Sha1);
3329 let base_hash = utils::calculate_object_hash(ObjectType::Blob, b"hello");
3330 let ofs_hash = utils::calculate_object_hash(ObjectType::Blob, b"hi there");
3331 let ref_hash = utils::calculate_object_hash(ObjectType::Blob, b"HELLO");
3332
3333 let mut pack_data = Vec::new();
3334 pack_data.extend_from_slice(b"PACK");
3335 pack_data.extend_from_slice(&2u32.to_be_bytes());
3336 pack_data.extend_from_slice(&3u32.to_be_bytes());
3337
3338 let base_offset = pack_data.len() as u32;
3339 pack_data.push(0x35);
3340 append_compressed(&mut pack_data, b"hello");
3341
3342 let ofs_offset = pack_data.len() as u32;
3343 pack_data.push(0x6b);
3344 pack_data.push((ofs_offset - base_offset) as u8);
3345 append_compressed(
3346 &mut pack_data,
3347 [b"\x05\x08\x08".as_ref(), b"hi there"].concat().as_slice(),
3348 );
3349
3350 let ref_offset = pack_data.len() as u32;
3351 pack_data.push(0x78);
3352 pack_data.extend_from_slice(base_hash.as_ref());
3353 append_compressed(
3354 &mut pack_data,
3355 [b"\x05\x05\x05".as_ref(), b"HELLO"].concat().as_slice(),
3356 );
3357
3358 let trailer = Sha1::digest(&pack_data);
3359 pack_data.extend_from_slice(&trailer);
3360
3361 let dir = tempdir().unwrap();
3362 let pack_path = dir.path().join("delta.pack");
3363 fs::write(&pack_path, &pack_data).unwrap();
3364 write_test_idx(
3365 &pack_path,
3366 vec![
3367 (base_hash, base_offset),
3368 (ofs_hash, ofs_offset),
3369 (ref_hash, ref_offset),
3370 ],
3371 );
3372
3373 let mut pack = Pack::new(Some(1), None, Some(dir.path().join("tmp")), true);
3374 pack.decode_file_without_callback(&pack_path, None::<fn(ObjectHash)>)
3375 .unwrap();
3376
3377 assert_eq!(pack.number, 3);
3378 assert_eq!(pack.signature.to_string(), hex::encode(trailer));
3379 }
3380
3381 #[test]
3382 fn test_rebuild_delta_literal_instruction() {
3383 let _guard = set_hash_kind_for_test(HashKind::Sha1);
3384 let base = Arc::new(CacheObject::new_for_undeltified(
3385 ObjectType::Blob,
3386 b"hello".to_vec(),
3387 12,
3388 0,
3389 ));
3390 let delta = CacheObject {
3391 info: CacheObjectInfo::OffsetDelta(12, 8),
3392 offset: 20,
3393 crc32: 0,
3394 data_decompressed: [b"\x05\x08\x08".as_ref(), b"hi there"].concat(),
3395 mem_recorder: None,
3396 is_delta_in_pack: true,
3397 known_hash: None,
3398 };
3399
3400 let rebuilt = Pack::rebuild_delta(delta, base).unwrap();
3401
3402 assert_eq!(rebuilt.object_type(), ObjectType::Blob);
3403 assert_eq!(rebuilt.data_decompressed, b"hi there");
3404 }
3405
3406 #[test]
3407 fn test_rebuild_delta_malformed_stream_returns_error() {
3408 let _guard = set_hash_kind_for_test(HashKind::Sha1);
3409 let base = Arc::new(CacheObject::new_for_undeltified(
3410 ObjectType::Blob,
3411 b"hello".to_vec(),
3412 12,
3413 0,
3414 ));
3415 let delta = CacheObject {
3416 info: CacheObjectInfo::OffsetDelta(12, 0),
3417 offset: 20,
3418 crc32: 0,
3419 data_decompressed: Vec::new(),
3420 mem_recorder: None,
3421 is_delta_in_pack: true,
3422 known_hash: None,
3423 };
3424
3425 let result = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
3426 Pack::rebuild_delta(delta, base)
3427 }));
3428
3429 assert!(result.is_ok(), "malformed delta should not panic");
3430 let result = result.unwrap();
3431 assert!(
3432 matches!(result, Err(crate::errors::GitError::DeltaObjectError(_))),
3433 "unexpected decode result: {result:?}"
3434 );
3435 }
3436
3437 #[test]
3438 fn test_pack_decode_malformed_delta_returns_error_without_panic() {
3439 let _guard = set_hash_kind_for_test(HashKind::Sha1);
3440 let mut pack_data = Vec::new();
3441 pack_data.extend_from_slice(b"PACK");
3442 pack_data.extend_from_slice(&2u32.to_be_bytes());
3443 pack_data.extend_from_slice(&2u32.to_be_bytes());
3444
3445 let base_offset = pack_data.len();
3446 pack_data.push(0x31);
3447 append_compressed(&mut pack_data, b"a");
3448
3449 let delta_offset = pack_data.len();
3450 let delta_payload = vec![0x01, 0x02, b'b'];
3451 pack_data.push(0x63);
3452 pack_data.push((delta_offset - base_offset) as u8);
3453 append_compressed(&mut pack_data, &delta_payload);
3454
3455 let trailer = Sha1::digest(&pack_data);
3456 pack_data.extend_from_slice(&trailer);
3457
3458 let result = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
3459 let mut reader = Cursor::new(pack_data);
3460 let mut pack = Pack::new(Some(1), None, None, true);
3461 pack.decode(&mut reader, |_| {}, None::<fn(ObjectHash)>)
3462 }));
3463
3464 assert!(result.is_ok(), "malformed pack delta should not panic");
3465 let result = result.unwrap();
3466 assert!(
3467 matches!(result, Err(crate::errors::GitError::DeltaObjectError(_))),
3468 "unexpected decode result: {result:?}"
3469 );
3470 }
3471
3472 #[test]
3473 #[cfg(target_pointer_width = "32")]
3474 fn test_pack_new_mem_limit_no_overflow_32bit() {
3475 let mem_limit = 1_200_000_000usize;
3479 let (_tmp_dir, tmp) = pack_test_tmp();
3480 let result = std::panic::catch_unwind(|| {
3481 let _p = Pack::new(Some(1), Some(mem_limit), Some(tmp), true);
3482 });
3483 assert!(result.is_ok(), "Pack::new should not panic on 32-bit");
3484 }
3485
3486 fn run_decode_no_delta(filename: &str, kind: HashKind) {
3488 let _guard = set_hash_kind_for_test(kind);
3489 let (source, _dl_guard) = download_pack_file(filename);
3490
3491 let (_tmp_dir, tmp) = pack_test_tmp();
3492
3493 let f = fs::File::open(source).unwrap();
3494 let mut buffered = BufReader::new(f);
3495 let mut p = Pack::new(None, Some(1024 * 1024 * 20), Some(tmp), true);
3496 p.decode(&mut buffered, |_| {}, None::<fn(ObjectHash)>)
3497 .unwrap();
3498 }
3499 #[test]
3500 fn test_pack_decode_without_delta() {
3501 run_decode_no_delta("small-sha1.pack", HashKind::Sha1);
3502 run_decode_no_delta("small-sha256.pack", HashKind::Sha256);
3503 }
3504
3505 fn run_decode_with_ref_delta(filename: &str, kind: HashKind) {
3507 let _guard = set_hash_kind_for_test(kind);
3508 init_logger();
3509
3510 let (source, _dl_guard) = download_pack_file(filename);
3511
3512 let (_tmp_dir, tmp) = pack_test_tmp();
3513
3514 let f = fs::File::open(source).unwrap();
3515 let mut buffered = BufReader::new(f);
3516 let mut p = Pack::new(None, Some(1024 * 1024 * 20), Some(tmp), true);
3517 p.decode(&mut buffered, |_| {}, None::<fn(ObjectHash)>)
3518 .unwrap();
3519 }
3520 #[test]
3521 fn test_pack_decode_with_ref_delta() {
3522 run_decode_with_ref_delta("ref-delta-sha1.pack", HashKind::Sha1);
3523 run_decode_with_ref_delta("ref-delta-sha256.pack", HashKind::Sha256);
3524 }
3525
3526 fn run_decode_no_mem_limit(filename: &str, kind: HashKind) {
3528 let _guard = set_hash_kind_for_test(kind);
3529 let (source, _dl_guard) = download_pack_file(filename);
3530
3531 let (_tmp_dir, tmp) = pack_test_tmp();
3532
3533 let f = fs::File::open(source).unwrap();
3534 let mut buffered = BufReader::new(f);
3535 let mut p = Pack::new(None, None, Some(tmp), true);
3536 p.decode(&mut buffered, |_| {}, None::<fn(ObjectHash)>)
3537 .unwrap();
3538 }
3539 #[test]
3540 fn test_pack_decode_no_mem_limit() {
3541 run_decode_no_mem_limit("small-sha1.pack", HashKind::Sha1);
3542 run_decode_no_mem_limit("small-sha256.pack", HashKind::Sha256);
3543 }
3544
3545 async fn run_decode_large_with_delta(filename: &str, kind: HashKind) {
3547 let _guard = set_hash_kind_for_test(kind);
3548 init_logger();
3549 let (source, _dl_guard) = download_pack_file(filename);
3550
3551 let (_tmp_dir, tmp) = pack_test_tmp();
3552
3553 let f = fs::File::open(source).unwrap();
3554 let mut buffered = BufReader::new(f);
3555 let mut p = Pack::new(
3556 Some(4),
3557 Some(LARGE_PACK_TEST_MEM_LIMIT),
3558 Some(tmp.clone()),
3559 true,
3560 );
3561 let rt = p.decode(
3562 &mut buffered,
3563 |_obj| {
3564 },
3566 None::<fn(ObjectHash)>,
3567 );
3568 if let Err(e) = rt {
3569 let _ = fs::remove_dir_all(&tmp);
3570 panic!("Error: {e:?}");
3571 }
3572 }
3573 #[cfg_attr(coverage, ignore)]
3574 #[ignore = "requires large remote pack fixture"]
3575 #[tokio::test]
3576 async fn test_pack_decode_with_large_file_with_delta_without_ref() {
3577 run_decode_large_with_delta("medium-sha1.pack", HashKind::Sha1).await;
3578 run_decode_large_with_delta("medium-sha256.pack", HashKind::Sha256).await;
3579 } async fn run_decode_large_stream(filename: &str, kind: HashKind) {
3583 let _guard = set_hash_kind_for_test(kind);
3584 init_logger();
3585 let (source, _dl_guard) = download_pack_file(filename);
3586
3587 let (_tmp_dir, tmp) = pack_test_tmp();
3588 let f = tokio::fs::File::open(source).await.unwrap();
3589 let stream = ReaderStream::new(f).map_err(axum::Error::new);
3590 let p = Pack::new(
3591 Some(4),
3592 Some(LARGE_PACK_TEST_MEM_LIMIT),
3593 Some(tmp.clone()),
3594 true,
3595 );
3596
3597 let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel();
3598 let handle = tokio::spawn(async move { p.decode_stream(stream, tx, None).await });
3599 let count = Arc::new(AtomicUsize::new(0));
3600 let count_c = count.clone();
3601 let consume = tokio::spawn(async move {
3603 let mut cnt = 0;
3604 while let Some(_entry) = rx.recv().await {
3605 cnt += 1;
3606 }
3607 tracing::info!("Received: {}", cnt);
3608 count_c.store(cnt, Ordering::Release);
3609 });
3610 let p = handle.await.unwrap();
3611 consume.await.unwrap();
3612 assert_eq!(count.load(Ordering::Acquire), p.number);
3613 assert_eq!(p.number, 35031);
3614 }
3615 #[cfg_attr(coverage, ignore)]
3616 #[ignore = "requires large remote pack fixture"]
3617 #[tokio::test]
3618 async fn test_decode_large_file_stream() {
3619 run_decode_large_stream("medium-sha1.pack", HashKind::Sha1).await;
3620 run_decode_large_stream("medium-sha256.pack", HashKind::Sha256).await;
3621 }
3622
3623 async fn run_decode_large_file_async(filename: &str, kind: HashKind) {
3625 let _guard = set_hash_kind_for_test(kind);
3626 let (source, _dl_guard) = download_pack_file(filename);
3627
3628 let (_tmp_dir, tmp) = pack_test_tmp();
3629 let f = fs::File::open(source).unwrap();
3630 let buffered = BufReader::new(f);
3631 let p = Pack::new(
3632 Some(4),
3633 Some(LARGE_PACK_TEST_MEM_LIMIT),
3634 Some(tmp.clone()),
3635 true,
3636 );
3637
3638 let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel();
3639 let handle = p.decode_async(buffered, tx); let mut cnt = 0;
3641 while let Some(_entry) = rx.recv().await {
3642 cnt += 1; }
3644 let p = handle.join().unwrap();
3645 assert_eq!(cnt, p.number);
3646 }
3647 #[cfg_attr(coverage, ignore)]
3648 #[ignore = "requires large remote pack fixture"]
3649 #[tokio::test]
3650 async fn test_decode_large_file_async() {
3651 run_decode_large_file_async("medium-sha1.pack", HashKind::Sha1).await;
3652 run_decode_large_file_async("medium-sha256.pack", HashKind::Sha256).await;
3653 }
3654
3655 fn run_decode_with_delta_no_ref(filename: &str, kind: HashKind) {
3657 let _guard = set_hash_kind_for_test(kind);
3658 let (source, _dl_guard) = download_pack_file(filename);
3659
3660 let (_tmp_dir, tmp) = pack_test_tmp();
3661
3662 let f = fs::File::open(source).unwrap();
3663 let mut buffered = BufReader::new(f);
3664 let mut p = Pack::new(None, Some(LARGE_PACK_TEST_MEM_LIMIT), Some(tmp), true);
3665 p.decode(&mut buffered, |_| {}, None::<fn(ObjectHash)>)
3666 .unwrap();
3667 }
3668 #[cfg_attr(coverage, ignore)]
3669 #[ignore = "requires large remote pack fixture"]
3670 #[test]
3671 fn test_pack_decode_with_delta_without_ref() {
3672 run_decode_with_delta_no_ref("medium-sha1.pack", HashKind::Sha1);
3673 run_decode_with_delta_no_ref("medium-sha256.pack", HashKind::Sha256);
3674 }
3675
3676 #[cfg_attr(coverage, ignore)]
3677 #[ignore = "requires large remote pack fixture"]
3678 #[test] fn test_pack_decode_multi_task_with_large_file_with_delta_without_ref() {
3680 let rt = tokio::runtime::Builder::new_current_thread()
3681 .enable_all()
3682 .build()
3683 .unwrap();
3684 rt.block_on(async move {
3685 for (kind, filename) in [
3687 (HashKind::Sha1, "medium-sha1.pack"),
3688 (HashKind::Sha256, "medium-sha256.pack"),
3689 ] {
3690 let f1 = run_decode_large_with_delta(filename, kind);
3691 let f2 = run_decode_large_with_delta(filename, kind);
3692 let _ = futures::future::join(f1, f2).await;
3693 }
3694 });
3695 }
3696
3697 #[test]
3709 fn test_stats_pack_small_sha1() {
3710 let _guard = set_hash_kind_for_test(HashKind::Sha1);
3711 let (source, _dl_guard) = download_pack_file("small-sha1.pack");
3712
3713 let stats = Pack::stats_pack(source).expect("stats_pack should succeed");
3714
3715 eprintln!(
3716 "small-sha1 stats: total={}, commits={}, trees={}, blobs={}, tags={}, deltas={}",
3717 stats.total, stats.commits, stats.trees, stats.blobs, stats.tags, stats.deltas
3718 );
3719
3720 let sum = stats.commits + stats.trees + stats.blobs + stats.tags + stats.deltas;
3721 assert_eq!(
3722 sum, stats.total,
3723 "per-type counts should sum to total ({} vs {})",
3724 sum, stats.total
3725 );
3726
3727 assert!(stats.commits > 0, "expected at least one commit");
3728 assert!(stats.blobs > 0, "expected at least one blob");
3729 }
3730
3731 #[test]
3736 fn test_stats_pack_medium_sha1_has_deltas() {
3737 let _guard = set_hash_kind_for_test(HashKind::Sha1);
3738 let (source, _dl_guard) = download_pack_file("medium-sha1.pack");
3739
3740 let stats = Pack::stats_pack(source).expect("stats_pack should succeed on medium pack");
3741
3742 eprintln!(
3743 "medium-sha1 stats: total={}, commits={}, trees={}, blobs={}, tags={}, deltas={}",
3744 stats.total, stats.commits, stats.trees, stats.blobs, stats.tags, stats.deltas
3745 );
3746
3747 let sum = stats.commits + stats.trees + stats.blobs + stats.tags + stats.deltas;
3748 assert_eq!(sum, stats.total, "per-type counts must equal total");
3749
3750 assert!(
3751 stats.deltas > 0,
3752 "expected delta objects in medium-sha1 pack"
3753 );
3754
3755 assert!(stats.total > 1000, "expected a sizeable medium pack");
3756 }
3757
3758 #[test]
3762 fn test_stats_pack_file_not_found() {
3763 let result = Pack::stats_pack(PathBuf::from("/nonexistent/path/to/fake.pack"));
3764 assert!(
3765 result.is_err(),
3766 "stats_pack should return Err for a missing file"
3767 );
3768 }
3769
3770 #[test]
3775 fn test_stats_pack_invalid_pack_magic() {
3776 use std::io::Write;
3777
3778 use tempfile::NamedTempFile;
3779
3780 let mut tmp = NamedTempFile::new().expect("create temp file");
3781 tmp.write_all(b"FAKE\x00\x00\x00\x02\x00\x00\x00\x05")
3782 .expect("write temp bytes");
3783 let path = tmp.path().to_path_buf();
3784
3785 let result = Pack::stats_pack(path);
3786 assert!(
3787 result.is_err(),
3788 "stats_pack should return Err for invalid pack magic"
3789 );
3790 }
3791}