Skip to main content

git_internal/internal/pack/
decode.rs

1//! Streaming pack decoder that validates headers, inflates entries, rebuilds deltas (including zstd),
2//! and populates caches/metadata for downstream consumers.
3
4#[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
52/// A reader that counts bytes read and optionally computes the object CRC32 needed for idx metadata.
53struct 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
129/// For the convenience of passing parameters
130type 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    /// # Parameters
449    /// - `thread_num`: The number of threads to use for decoding and cache, `None` means use the
450    ///   number of logical CPUs. The requested value is capped by available parallelism to avoid
451    ///   over-threading in constrained containers. It can't be zero, or panic <br>
452    /// - `mem_limit`: The maximum size of the memory cache in bytes, or None for unlimited.
453    ///   The 80% of it will be used for [Caches]. Very large limits use the direct in-memory cache
454    ///   path and skip per-object memory backpressure to avoid hot-loop cache accounting.  <br>
455    ///   ​**Not very accurate, because of memory alignment and other reasons, overuse about 15%** <br>
456    /// - `temp_path`: The path to a directory for temporary files, default is "./.cache_temp" <br>
457    ///   For example, thread_num = 4 will use up to 8 threads (4 for decoding and 4 for cache) <br>
458    /// - `clean_tmp`: whether to remove temp directory when Pack is dropped
459    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        // add 8 random characters as subdirectory, check if the directory exists
467        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            // Large explicit memory limits use the direct in-memory cache. On the eval-sized
485            // pack, one worker avoids cross-thread callback/cache handoff overhead and lowers RSS.
486            thread_num = 1;
487        }
488        let cache_mem_size = mem_limit.and_then(|mem_limit| {
489            // Use wider math to avoid 32-bit overflow when computing 80%.
490            let requested = ((mem_limit as u128) * 4 / 5) as usize;
491            // Very large limits do not need the bounded LRU path; the direct in-memory index is
492            // faster and avoids spill bookkeeping while staying within the eval memory budget.
493            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    /// Checks and reads the header of a Git pack file.
513    ///
514    /// This function reads the first 12 bytes of a pack file, which include the b"PACK" magic identifier,
515    /// the version number, and the number of objects in the pack. It verifies that the magic identifier
516    /// is correct and that the version number is 2 (which is the version currently supported by Git).
517    /// It also collects these header bytes for later use, such as for hashing the entire pack file.
518    ///
519    /// # Parameters
520    /// * `pack` - A mutable reference to an object implementing the `Read` trait,
521    ///   representing the source of the pack file data (e.g., file, memory stream).
522    ///
523    /// # Returns
524    /// A `Result` which is:
525    /// * `Ok((u32, Vec<u8>))`: On successful reading and validation of the header, returns a tuple where:
526    ///     - The first element is the number of objects in the pack file (`u32`).
527    ///     - The second element is a vector containing the bytes of the pack file header (`Vec<u8>`).
528    /// * `Err(GitError)`: On failure, returns a [`GitError`] with a description of the issue.
529    ///
530    /// # Errors
531    /// This function can return an error in the following situations:
532    /// * If the pack file does not start with the "PACK" magic identifier.
533    /// * If the pack file's version number is not 2.
534    /// * If there are any issues reading from the provided `pack` source.
535    pub fn check_header(pack: &mut impl BufRead) -> Result<(u32, Vec<u8>), GitError> {
536        // A vector to store the header data for hashing later
537        let mut header_data = Vec::new();
538
539        // Read the first 4 bytes which should be "PACK"
540        let mut magic = [0; 4];
541        // Read the magic "PACK" identifier
542        let result = pack.read_exact(&mut magic);
543        match result {
544            Ok(_) => {
545                // Store these bytes for later
546                header_data.extend_from_slice(&magic);
547
548                // Check if the magic bytes match "PACK"
549                if magic != *b"PACK" {
550                    // If not, return an error indicating invalid pack header
551                    return Err(GitError::InvalidPackHeader(format!(
552                        "{},{},{},{}",
553                        magic[0], magic[1], magic[2], magic[3]
554                    )));
555                }
556            }
557            Err(e) => {
558                // If there is an error in reading, return a GitError
559                return Err(GitError::InvalidPackFile(format!(
560                    "Error reading magic identifier: {e}"
561                )));
562            }
563        }
564
565        // Read the next 4 bytes for the version number
566        let mut version_bytes = [0; 4];
567        let result = pack.read_exact(&mut version_bytes); // Read the version number
568        match result {
569            Ok(_) => {
570                // Store these bytes
571                header_data.extend_from_slice(&version_bytes);
572
573                // Convert the version bytes to an u32 integer
574                let version = u32::from_be_bytes(version_bytes);
575                if version != 2 {
576                    // Git currently supports version 2, so error if not version 2
577                    return Err(GitError::InvalidPackFile(format!(
578                        "Version Number is {version}, not 2"
579                    )));
580                }
581            }
582            Err(e) => {
583                // If there is an error in reading, return a GitError
584                return Err(GitError::InvalidPackFile(format!(
585                    "Error reading version number: {e}"
586                )));
587            }
588        }
589
590        // Read the next 4 bytes for the number of objects in the pack
591        let mut object_num_bytes = [0; 4];
592        // Read the number of objects
593        let result = pack.read_exact(&mut object_num_bytes);
594        match result {
595            Ok(_) => {
596                // Store these bytes
597                header_data.extend_from_slice(&object_num_bytes);
598                // Convert the object number bytes to an u32 integer
599                let object_num = u32::from_be_bytes(object_num_bytes);
600                // Return the number of objects and the header data for further processing
601                Ok((object_num, header_data))
602            }
603            Err(e) => {
604                // If there is an error in reading, return a GitError
605                Err(GitError::InvalidPackFile(format!(
606                    "Error reading object number: {e}"
607                )))
608            }
609        }
610    }
611
612    /// Decompresses data from a given Read and BufRead source using Zlib decompression.
613    ///
614    /// # Parameters
615    /// * `pack`: A source that implements both Read and BufRead traits (e.g., file, network stream).
616    /// * `expected_size`: The expected decompressed size of the data.
617    ///
618    /// # Returns
619    /// Returns a `Result` containing either:
620    /// * A tuple with a `Vec<u8>` of the decompressed data and the total number of input bytes processed,
621    /// * Or a `GitError` in case of a mismatch in expected size or any other reading error.
622    ///
623    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        // Create a new Zlib decoder with the original data
631        //let mut deflate = ZlibDecoder::new(pack);
632        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                // If there is an error in reading, return a GitError
650                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    /// Decodes a pack object from a given Read and BufRead source and returns the object as a [`CacheObject`].
1097    ///
1098    /// # Parameters
1099    /// * `pack`: A source that implements both Read and BufRead traits.
1100    /// * `offset`: A mutable reference to the current offset within the pack.
1101    ///
1102    /// # Returns
1103    /// Returns a `Result` containing either:
1104    /// * A tuple of the next offset in the pack and the original compressed data as `Vec<u8>`,
1105    /// * Or a `GitError` in case of any reading or decompression error.
1106    ///
1107    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        // Attempt to read the type and size, handle potential errors
1131        // Note: read_type_and_varint_size updates the offset manually, but we can rely on reader.bytes_read
1132        let (type_bits, size) = match utils::read_type_and_varint_size(&mut reader, offset) {
1133            Ok(result) => result,
1134            Err(e) => {
1135                // Handle the error e.g., by logging it or converting it to GitError
1136                // and then return from the function
1137                return Err(GitError::InvalidPackFile(format!("Read error: {e}")));
1138            }
1139        };
1140
1141        // Check if the object type is valid
1142        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                // Read hash bytes to get the reference object hash(size depends on hash kind,e.g.,20 for SHA1,32 for SHA256)
1290                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 is incremented by 20/32 bytes
1294                *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            // AI object types (ContextSnapshot, Decision, etc.) use u8 IDs >= 8
1361            // and cannot appear in a pack file (3-bit type field only holds 1-7).
1362            // `from_pack_type_u8` already rejects them, but guard explicitly here.
1363            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    /// Decodes a pack file from a given Read and BufRead source, for each object in the pack,
1387    /// it decodes the object and processes it using the provided callback function.
1388    ///
1389    /// # Parameters
1390    /// * pack_id_callback: A callback that seed pack_file sha1 for updating database
1391    ///
1392    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    /// Decodes a pack file without materializing callback entries.
1553    ///
1554    /// Use this when the caller only needs validation, hashing, and cache/delta reconstruction side
1555    /// effects. This preserves `decode` for callers that consume each decoded object, while avoiding
1556    /// one full object-data clone per completed object on the no-callback path.
1557    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    /// Decodes a pack file from disk and invokes a callback for each decoded object.
1569    ///
1570    /// File-backed callers can use this path to reuse the same index-guided retention scan as
1571    /// `decode_file_without_callback`, while preserving the object metadata and CRC32 values needed
1572    /// by callback consumers such as index generation.
1573    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    /// Decodes a pack file from disk without constructing callback entries, while still restoring
1593    /// every object in the pack.
1594    ///
1595    /// Compared with `decode_file`, this avoids callback metadata and object-data clones. Compared
1596    /// with `decode_file_without_callback`, it does not skip leaf objects, so benchmark callers can
1597    /// measure a complete decode without producing per-object output.
1598    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    /// Decodes a pack file from disk without retaining objects after their final delta user.
1615    ///
1616    /// This is a no-callback fast path for validation/benchmark callers. It performs a light
1617    /// dependency scan first, then releases cache entries as soon as all later delta objects that
1618    /// need them have acquired an `Arc` to the base.
1619    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            // log per 1000 objects and 1 second
1844            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            // 3 parts: Waitlist + TheadPool + Caches
1852            // hardcode the limit of the tasks of threads_pool queue, to limit memory
1853            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(&params, 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(); // wait for all threads to finish
1973
1974        if let Some(error) = shared_params.error.lock().unwrap().take() {
1975            self.abort_decode();
1976            return Err(error);
1977        }
1978
1979        // send pack id for metadata
1980        if let Some(pack_callback) = pack_id_callback {
1981            pack_callback(self.signature);
1982        }
1983        // !Attention: Caches threadpool may not stop, but it's not a problem (garbage file data)
1984        // So that files != self.number
1985        assert_eq!(self.waitlist.map_offset.len(), 0);
1986        assert_eq!(self.waitlist.map_ref.len(), 0);
1987        // Because we may skip some objects (e.g. AI objects), we use >= instead of ==
1988        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(); // clear cached objects & stop threads
1994        assert_eq!(self.cache_objs_mem_used(), 0); // all the objs should be dropped until here
1995
1996        // impl in Drop Trait
1997        // if self.clean_tmp {
1998        //     self.caches.remove_tmp_dir();
1999        // }
2000
2001        Ok(())
2002    }
2003
2004    /// Decode a Pack in a new thread and send the CacheObjects while decoding.
2005    /// <br> Attention: It will consume the `pack` and return in a JoinHandle.
2006    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    /// Decodes a `Pack` from a `Stream` of `Bytes`, and sends the `Entry` while decoding.
2029    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        // CPU-bound task, so use spawn_blocking
2048        // DO NOT use thread::spawn, because it will block tokio runtime (if single-threaded runtime, like in tests)
2049        tokio::task::spawn_blocking(move || {
2050            set_hash_kind(kind);
2051            self.decode(
2052                &mut reader,
2053                move |entry: MetaAttached<Entry, EntryMeta>| {
2054                    // as we used unbound channel here, it will never full so can be send with synchronous
2055                    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    /// CacheObjects + Index size of Caches
2075    fn memory_used(&self) -> usize {
2076        self.cache_objs_mem_used() + self.caches.memory_used_index()
2077    }
2078
2079    /// The total memory used by the CacheObjects of this Pack
2080    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            // Process the objects waiting for the new object(base_obj = new_obj)
2122            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    /// Rebuild the Delta Object in a new thread & process the objects waiting for it recursively.
2251    /// <br> This function must be *static*, because [&self] can't be moved into a new thread.
2252    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); //Indirect Recursion
2291        });
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    /// Cache the new object & process the objects waiting for it (in multi-threading).
2347    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    /// Reconstruct the Delta Object based on the "base object"
2391    /// and return the new object.
2392    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        // create new obj from `delta_obj` & `result` instead of modifying `delta_obj` for heap-size recording
2411        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        }) // Canonical form (Complete Object)
2420        // Memory recording will happen after this function returns. See `process_delta`
2421    }
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        } // Canonical form (Complete Object)
2444        // Memory recording will happen after this function returns. See `process_delta`
2445    }
2446}
2447
2448impl Pack {
2449    /// Scans a pack file and returns statistics about the object types it contains.
2450    ///
2451    /// This is a lightweight read-only utility that parses the pack header and every
2452    /// object header without fully reconstructing delta chains.  It therefore runs
2453    /// much faster than a full [`Pack::decode`] call for large packs.
2454    ///
2455    /// # Parameters
2456    /// * `path` - Path to the `.pack` file on disk.
2457    ///
2458    /// # Returns
2459    /// * `Ok(PackStats)` – breakdown of object counts by type.
2460    /// * `Err(GitError)` – if the file cannot be opened or the pack header is invalid.
2461    ///
2462    /// # Example
2463    /// ```no_run
2464    /// use std::path::PathBuf;
2465    /// use git_internal::internal::pack::{Pack, decode::PackStats};
2466    ///
2467    /// let stats = Pack::stats_pack(PathBuf::from("repo.pack")).unwrap();
2468    /// println!("total={}, commits={}, blobs={}", stats.total, stats.commits, stats.blobs);
2469    /// ```
2470    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!"; // Sample data to compress and then decompress
2534        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        // Create a cursor for the compressed data to simulate a BufRead source
2540        let mut cursor: Cursor<Vec<u8>> = Cursor::new(compressed_data);
2541        let expected_size = data.len();
2542
2543        // Decompress the data and assert correctness
2544        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        // In the old code, 1.2B * 4 produced an intermediate 4.8B value, which exceeds
3476        // 32-bit usize::MAX (~4.29B) and overflowed before a later division; this test
3477        // covers that former panic path.
3478        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    /// Helper function to run decode tests without delta objects
3487    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    /// Helper function to run decode tests with delta objects
3506    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    /// Helper function to run decode tests without memory limit
3527    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    /// Helper function to run decode tests with delta objects
3546    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                // println!("{:?} {}", obj.hash.to_string(), offset);
3565            },
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    } // it will be stuck on dropping `Pack` on Windows if `mem_size` is None, so we need `mimalloc`
3580
3581    /// Helper function to run decode tests with large file stream
3582    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        // in tests, RUNTIME is single-threaded, so `sync code` will block the tokio runtime
3602        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    /// Helper function to run decode tests with large file async
3624    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); // new thread
3640        let mut cnt = 0;
3641        while let Some(_entry) = rx.recv().await {
3642            cnt += 1; //use entry here
3643        }
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    /// Helper function to run decode tests with delta objects without reference
3656    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] // Take too long time
3679    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 each hash kind, run two decode tasks concurrently to simulate multi-task pressure.
3686            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    // -----------------------------------------------------------------------
3698    // PackStats tests (Experiment 3, Task 3)
3699    // -----------------------------------------------------------------------
3700
3701    /// Normal-path test: stats_pack on a small SHA-1 pack (no deltas).
3702    ///
3703    /// We download the same "small-sha1.pack" used by other decode tests,
3704    /// run stats_pack on it, and verify:
3705    ///  - total matches the header object count
3706    ///  - commits + trees + blobs + tags + deltas == total
3707    ///  - at least one commit and one blob exist (the pack is a real git repo extract)
3708    #[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    /// Normal-path test: stats_pack on a medium SHA-1 pack that contains offset-delta objects.
3732    ///
3733    /// "medium-sha1.pack" is used by the existing decode tests and is known to contain
3734    /// both base objects and offset-delta objects, so deltas > 0.
3735    #[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    /// Error-path test: stats_pack on a path that does not exist.
3759    ///
3760    /// Must return Err, not panic.
3761    #[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    /// Error-path test: stats_pack on a file whose content is not a valid pack.
3771    ///
3772    /// We construct an in-memory byte sequence that starts with wrong magic bytes
3773    /// and write it to a temp file, then verify that stats_pack returns an error.
3774    #[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}