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