git_internal/internal/pack/

decode.rs

//! Streaming pack decoder that validates headers, inflates entries, rebuilds deltas (including zstd),
//! and populates caches/metadata for downstream consumers.

use std::{
    io::{self, BufRead, Cursor, ErrorKind, Read},
    path::PathBuf,
    sync::{
        Arc,
        atomic::{AtomicUsize, Ordering},
    },
    thread::{self, JoinHandle},
    time::Instant,
};

use axum::Error;
use bytes::Bytes;
use flate2::bufread::ZlibDecoder;
use futures_util::{Stream, StreamExt};
use threadpool::ThreadPool;
use tokio::sync::mpsc::UnboundedSender;
use uuid::Uuid;

use super::cache_object::CacheObjectInfo;
use crate::{
    errors::GitError,
    hash::{ObjectHash, get_hash_kind, set_hash_kind},
    internal::{
        metadata::{EntryMeta, MetaAttached},
        object::types::ObjectType,
        pack::{
            DEFAULT_TMP_DIR, Pack,
            cache::{_Cache, Caches},
            cache_object::{CacheObject, MemSizeRecorder},
            channel_reader::StreamBufReader,
            entry::Entry,
            utils,
            waitlist::Waitlist,
            wrapper::Wrapper,
        },
    },
    utils::CountingReader,
    zstdelta,
};

/// A reader that counts bytes read and computes CRC32 checksum.
/// which is used to verify the integrity of decompressed data.
struct CrcCountingReader<'a, R> {
    inner: R,
    bytes_read: u64,
    crc: &'a mut crc32fast::Hasher,
}
impl<R: Read> Read for CrcCountingReader<'_, R> {
    fn read(&mut self, buf: &mut [u8]) -> io::Result<usize> {
        let n = self.inner.read(buf)?;
        self.bytes_read += n as u64;
        self.crc.update(&buf[..n]);
        Ok(n)
    }
}
impl<R: BufRead> BufRead for CrcCountingReader<'_, R> {
    fn fill_buf(&mut self) -> io::Result<&[u8]> {
        self.inner.fill_buf()
    }
    fn consume(&mut self, amt: usize) {
        let buf = self.inner.fill_buf().unwrap_or(&[]);
        self.crc.update(&buf[..amt.min(buf.len())]);
        self.bytes_read += amt as u64;
        self.inner.consume(amt);
    }
}

/// For the convenience of passing parameters
struct SharedParams {
    pub pool: Arc<ThreadPool>,
    pub waitlist: Arc<Waitlist>,
    pub caches: Arc<Caches>,
    pub cache_objs_mem_size: Arc<AtomicUsize>,
    pub callback: Arc<dyn Fn(MetaAttached<Entry, EntryMeta>) + Sync + Send>,
}

impl Drop for Pack {
    fn drop(&mut self) {
        if self.clean_tmp {
            self.caches.remove_tmp_dir();
        }
    }
}

impl Pack {
    /// # Parameters
    /// - `thread_num`: The number of threads to use for decoding and cache, `None` mean use the number of logical CPUs.
    ///   It can't be zero, or panic <br>
    /// - `mem_limit`: The maximum size of the memory cache in bytes, or None for unlimited.
    ///   The 80% of it will be used for [Caches]  <br>
    ///   ​**Not very accurate, because of memory alignment and other reasons, overuse about 15%** <br>
    /// - `temp_path`: The path to a directory for temporary files, default is "./.cache_temp" <br>
    ///   For example, thread_num = 4 will use up to 8 threads (4 for decoding and 4 for cache) <br>
    /// - `clean_tmp`: whether to remove temp directory when Pack is dropped
    pub fn new(
        thread_num: Option<usize>,
        mem_limit: Option<usize>,
        temp_path: Option<PathBuf>,
        clean_tmp: bool,
    ) -> Self {
        let mut temp_path = temp_path.unwrap_or(PathBuf::from(DEFAULT_TMP_DIR));
        // add 8 random characters as subdirectory, check if the directory exists
        loop {
            let sub_dir = Uuid::new_v4().to_string()[..8].to_string();
            temp_path.push(sub_dir);
            if !temp_path.exists() {
                break;
            }
            temp_path.pop();
        }
        let thread_num = thread_num.unwrap_or_else(num_cpus::get);
        let cache_mem_size = mem_limit.map(|mem_limit| mem_limit * 4 / 5);
        Pack {
            number: 0,
            signature: ObjectHash::default(),
            objects: Vec::new(),
            pool: Arc::new(ThreadPool::new(thread_num)),
            waitlist: Arc::new(Waitlist::new()),
            caches: Arc::new(Caches::new(cache_mem_size, temp_path, thread_num)),
            mem_limit,
            cache_objs_mem: Arc::new(AtomicUsize::default()),
            clean_tmp,
        }
    }

    /// Checks and reads the header of a Git pack file.
    ///
    /// This function reads the first 12 bytes of a pack file, which include the b"PACK" magic identifier,
    /// the version number, and the number of objects in the pack. It verifies that the magic identifier
    /// is correct and that the version number is 2 (which is the version currently supported by Git).
    /// It also collects these header bytes for later use, such as for hashing the entire pack file.
    ///
    /// # Parameters
    /// * `pack` - A mutable reference to an object implementing the `Read` trait,
    ///   representing the source of the pack file data (e.g., file, memory stream).
    ///
    /// # Returns
    /// A `Result` which is:
    /// * `Ok((u32, Vec<u8>))`: On successful reading and validation of the header, returns a tuple where:
    ///     - The first element is the number of objects in the pack file (`u32`).
    ///     - The second element is a vector containing the bytes of the pack file header (`Vec<u8>`).
    /// * `Err(GitError)`: On failure, returns a [`GitError`] with a description of the issue.
    ///
    /// # Errors
    /// This function can return an error in the following situations:
    /// * If the pack file does not start with the "PACK" magic identifier.
    /// * If the pack file's version number is not 2.
    /// * If there are any issues reading from the provided `pack` source.
    pub fn check_header(pack: &mut impl BufRead) -> Result<(u32, Vec<u8>), GitError> {
        // A vector to store the header data for hashing later
        let mut header_data = Vec::new();

        // Read the first 4 bytes which should be "PACK"
        let mut magic = [0; 4];
        // Read the magic "PACK" identifier
        let result = pack.read_exact(&mut magic);
        match result {
            Ok(_) => {
                // Store these bytes for later
                header_data.extend_from_slice(&magic);

                // Check if the magic bytes match "PACK"
                if magic != *b"PACK" {
                    // If not, return an error indicating invalid pack header
                    return Err(GitError::InvalidPackHeader(format!(
                        "{},{},{},{}",
                        magic[0], magic[1], magic[2], magic[3]
                    )));
                }
            }
            Err(e) => {
                // If there is an error in reading, return a GitError
                return Err(GitError::InvalidPackFile(format!(
                    "Error reading magic identifier: {e}"
                )));
            }
        }

        // Read the next 4 bytes for the version number
        let mut version_bytes = [0; 4];
        let result = pack.read_exact(&mut version_bytes); // Read the version number
        match result {
            Ok(_) => {
                // Store these bytes
                header_data.extend_from_slice(&version_bytes);

                // Convert the version bytes to an u32 integer
                let version = u32::from_be_bytes(version_bytes);
                if version != 2 {
                    // Git currently supports version 2, so error if not version 2
                    return Err(GitError::InvalidPackFile(format!(
                        "Version Number is {version}, not 2"
                    )));
                }
            }
            Err(e) => {
                // If there is an error in reading, return a GitError
                return Err(GitError::InvalidPackFile(format!(
                    "Error reading version number: {e}"
                )));
            }
        }

        // Read the next 4 bytes for the number of objects in the pack
        let mut object_num_bytes = [0; 4];
        // Read the number of objects
        let result = pack.read_exact(&mut object_num_bytes);
        match result {
            Ok(_) => {
                // Store these bytes
                header_data.extend_from_slice(&object_num_bytes);
                // Convert the object number bytes to an u32 integer
                let object_num = u32::from_be_bytes(object_num_bytes);
                // Return the number of objects and the header data for further processing
                Ok((object_num, header_data))
            }
            Err(e) => {
                // If there is an error in reading, return a GitError
                Err(GitError::InvalidPackFile(format!(
                    "Error reading object number: {e}"
                )))
            }
        }
    }

    /// Decompresses data from a given Read and BufRead source using Zlib decompression.
    ///
    /// # Parameters
    /// * `pack`: A source that implements both Read and BufRead traits (e.g., file, network stream).
    /// * `expected_size`: The expected decompressed size of the data.
    ///
    /// # Returns
    /// Returns a `Result` containing either:
    /// * A tuple with a `Vec<u8>` of the decompressed data and the total number of input bytes processed,
    /// * Or a `GitError` in case of a mismatch in expected size or any other reading error.
    ///
    pub fn decompress_data(
        pack: &mut (impl BufRead + Send),
        expected_size: usize,
    ) -> Result<(Vec<u8>, usize), GitError> {
        // Create a buffer with the expected size for the decompressed data
        let mut buf = Vec::with_capacity(expected_size);

        let mut counting_reader = CountingReader::new(pack);
        // Create a new Zlib decoder with the original data
        //let mut deflate = ZlibDecoder::new(pack);
        let mut deflate = ZlibDecoder::new(&mut counting_reader);
        // Attempt to read data to the end of the buffer
        match deflate.read_to_end(&mut buf) {
            Ok(_) => {
                // Check if the length of the buffer matches the expected size
                if buf.len() != expected_size {
                    Err(GitError::InvalidPackFile(format!(
                        "The object size {} does not match the expected size {}",
                        buf.len(),
                        expected_size
                    )))
                } else {
                    // If everything is as expected, return the buffer, the original data, and the total number of input bytes processed
                    let actual_input_bytes = counting_reader.bytes_read as usize;
                    Ok((buf, actual_input_bytes))
                }
            }
            Err(e) => {
                // If there is an error in reading, return a GitError
                Err(GitError::InvalidPackFile(format!(
                    "Decompression error: {e}"
                )))
            }
        }
    }

    /// Decodes a pack object from a given Read and BufRead source and returns the object as a [`CacheObject`].
    ///
    /// # Parameters
    /// * `pack`: A source that implements both Read and BufRead traits.
    /// * `offset`: A mutable reference to the current offset within the pack.
    ///
    /// # Returns
    /// Returns a `Result` containing either:
    /// * A tuple of the next offset in the pack and the original compressed data as `Vec<u8>`,
    /// * Or a `GitError` in case of any reading or decompression error.
    ///
    pub fn decode_pack_object(
        pack: &mut (impl BufRead + Send),
        offset: &mut usize,
    ) -> Result<CacheObject, GitError> {
        let init_offset = *offset;
        let mut hasher = crc32fast::Hasher::new();
        let mut reader = CrcCountingReader {
            inner: pack,
            bytes_read: 0,
            crc: &mut hasher,
        };

        // Attempt to read the type and size, handle potential errors
        // Note: read_type_and_varint_size updates the offset manually, but we can rely on reader.bytes_read
        let (type_bits, size) = match utils::read_type_and_varint_size(&mut reader, offset) {
            Ok(result) => result,
            Err(e) => {
                // Handle the error e.g., by logging it or converting it to GitError
                // and then return from the function
                return Err(GitError::InvalidPackFile(format!("Read error: {e}")));
            }
        };

        // Check if the object type is valid
        let t = ObjectType::from_u8(type_bits)?;

        match t {
            ObjectType::Commit | ObjectType::Tree | ObjectType::Blob | ObjectType::Tag => {
                let (data, raw_size) = Pack::decompress_data(&mut reader, size)?;
                *offset += raw_size;
                let crc32 = hasher.finalize();
                Ok(CacheObject::new_for_undeltified(
                    t,
                    data,
                    init_offset,
                    crc32,
                ))
            }
            ObjectType::OffsetDelta | ObjectType::OffsetZstdelta => {
                let (delta_offset, bytes) = utils::read_offset_encoding(&mut reader).unwrap();
                *offset += bytes;

                let (data, raw_size) = Pack::decompress_data(&mut reader, size)?;
                *offset += raw_size;

                // Count the base object offset: the current offset - delta offset
                let base_offset = init_offset
                    .checked_sub(delta_offset as usize)
                    .ok_or_else(|| {
                        GitError::InvalidObjectInfo("Invalid OffsetDelta offset".to_string())
                    })
                    .unwrap();

                let mut reader = Cursor::new(&data);
                let (_, final_size) = utils::read_delta_object_size(&mut reader)?;

                let obj_info = match t {
                    ObjectType::OffsetDelta => {
                        CacheObjectInfo::OffsetDelta(base_offset, final_size)
                    }
                    ObjectType::OffsetZstdelta => {
                        CacheObjectInfo::OffsetZstdelta(base_offset, final_size)
                    }
                    _ => unreachable!(),
                };
                let crc32 = hasher.finalize();
                Ok(CacheObject {
                    info: obj_info,
                    offset: init_offset,
                    crc32,
                    data_decompressed: data,
                    mem_recorder: None,
                    is_delta_in_pack: true,
                })
            }
            ObjectType::HashDelta => {
                // Read hash bytes to get the reference object hash(size depends on hash kind,e.g.,20 for SHA1,32 for SHA256)
                let ref_sha = ObjectHash::from_stream(&mut reader).unwrap();
                // Offset is incremented by 20/32 bytes
                *offset += get_hash_kind().size();

                let (data, raw_size) = Pack::decompress_data(&mut reader, size)?;
                *offset += raw_size;

                let mut reader = Cursor::new(&data);
                let (_, final_size) = utils::read_delta_object_size(&mut reader)?;

                let crc32 = hasher.finalize();

                Ok(CacheObject {
                    info: CacheObjectInfo::HashDelta(ref_sha, final_size),
                    offset: init_offset,
                    crc32,
                    data_decompressed: data,
                    mem_recorder: None,
                    is_delta_in_pack: true,
                })
            }
        }
    }

    /// Decodes a pack file from a given Read and BufRead source, for each object in the pack,
    /// it decodes the object and processes it using the provided callback function.
    ///
    /// # Parameters
    /// * pack_id_callback: A callback that seed pack_file sha1 for updating database
    ///
    pub fn decode<F, C>(
        &mut self,
        pack: &mut (impl BufRead + Send),
        callback: F,
        pack_id_callback: Option<C>,
    ) -> Result<(), GitError>
    where
        F: Fn(MetaAttached<Entry, EntryMeta>) + Sync + Send + 'static,
        C: FnOnce(ObjectHash) + Send + 'static,
    {
        let time = Instant::now();
        let mut last_update_time = time.elapsed().as_millis();
        let log_info = |_i: usize, pack: &Pack| {
            tracing::info!(
                "time {:.2} s \t decode: {:?} \t dec-num: {} \t cah-num: {} \t Objs: {} MB \t CacheUsed: {} MB",
                time.elapsed().as_millis() as f64 / 1000.0,
                _i,
                pack.pool.queued_count(),
                pack.caches.queued_tasks(),
                pack.cache_objs_mem_used() / 1024 / 1024,
                pack.caches.memory_used() / 1024 / 1024
            );
        };
        let callback = Arc::new(callback);

        let caches = self.caches.clone();
        let mut reader = Wrapper::new(io::BufReader::new(pack));

        let result = Pack::check_header(&mut reader);
        match result {
            Ok((object_num, _)) => {
                self.number = object_num as usize;
            }
            Err(e) => {
                return Err(e);
            }
        }
        tracing::info!("The pack file has {} objects", self.number);
        let mut offset: usize = 12;
        let mut i = 0;
        while i < self.number {
            // log per 1000 objects and 1 second
            if i % 1000 == 0 {
                let time_now = time.elapsed().as_millis();
                if time_now - last_update_time > 1000 {
                    log_info(i, self);
                    last_update_time = time_now;
                }
            }
            // 3 parts: Waitlist + TheadPool + Caches
            // hardcode the limit of the tasks of threads_pool queue, to limit memory
            while self.pool.queued_count() > 2000
                || self
                    .mem_limit
                    .map(|limit| self.memory_used() > limit)
                    .unwrap_or(false)
            {
                thread::yield_now();
            }
            let r: Result<CacheObject, GitError> =
                Pack::decode_pack_object(&mut reader, &mut offset);
            match r {
                Ok(mut obj) => {
                    obj.set_mem_recorder(self.cache_objs_mem.clone());
                    obj.record_mem_size();

                    // Wrapper of Arc Params, for convenience to pass
                    let params = Arc::new(SharedParams {
                        pool: self.pool.clone(),
                        waitlist: self.waitlist.clone(),
                        caches: self.caches.clone(),
                        cache_objs_mem_size: self.cache_objs_mem.clone(),
                        callback: callback.clone(),
                    });

                    let caches = caches.clone();
                    let waitlist = self.waitlist.clone();
                    let kind = get_hash_kind();
                    self.pool.execute(move || {
                        set_hash_kind(kind);
                        match obj.info {
                            CacheObjectInfo::BaseObject(_, _) => {
                                Self::cache_obj_and_process_waitlist(params, obj);
                            }
                            CacheObjectInfo::OffsetDelta(base_offset, _)
                            | CacheObjectInfo::OffsetZstdelta(base_offset, _) => {
                                if let Some(base_obj) = caches.get_by_offset(base_offset) {
                                    Self::process_delta(params, obj, base_obj);
                                } else {
                                    // You can delete this 'if' block ↑, because there are Second check in 'else'
                                    // It will be more readable, but the performance will be slightly reduced
                                    waitlist.insert_offset(base_offset, obj);
                                    // Second check: prevent that the base_obj thread has finished before the waitlist insert
                                    if let Some(base_obj) = caches.get_by_offset(base_offset) {
                                        Self::process_waitlist(params, base_obj);
                                    }
                                }
                            }
                            CacheObjectInfo::HashDelta(base_ref, _) => {
                                if let Some(base_obj) = caches.get_by_hash(base_ref) {
                                    Self::process_delta(params, obj, base_obj);
                                } else {
                                    waitlist.insert_ref(base_ref, obj);
                                    if let Some(base_obj) = caches.get_by_hash(base_ref) {
                                        Self::process_waitlist(params, base_obj);
                                    }
                                }
                            }
                        }
                    });
                }
                Err(e) => {
                    return Err(e);
                }
            }
            i += 1;
        }
        log_info(i, self);
        let render_hash = reader.final_hash();
        self.signature = ObjectHash::from_stream(&mut reader).unwrap();

        if render_hash != self.signature {
            return Err(GitError::InvalidPackFile(format!(
                "The pack file hash {} does not match the trailer hash {}",
                render_hash, self.signature
            )));
        }

        let end = utils::is_eof(&mut reader);
        if !end {
            return Err(GitError::InvalidPackFile(
                "The pack file is not at the end".to_string(),
            ));
        }

        self.pool.join(); // wait for all threads to finish

        // send pack id for metadata
        if let Some(pack_callback) = pack_id_callback {
            pack_callback(self.signature);
        }
        // !Attention: Caches threadpool may not stop, but it's not a problem (garbage file data)
        // So that files != self.number
        assert_eq!(self.waitlist.map_offset.len(), 0);
        assert_eq!(self.waitlist.map_ref.len(), 0);
        assert_eq!(self.number, caches.total_inserted());
        tracing::info!(
            "The pack file has been decoded successfully, takes: [ {:?} ]",
            time.elapsed()
        );
        self.caches.clear(); // clear cached objects & stop threads
        assert_eq!(self.cache_objs_mem_used(), 0); // all the objs should be dropped until here

        // impl in Drop Trait
        // if self.clean_tmp {
        //     self.caches.remove_tmp_dir();
        // }

        Ok(())
    }

    /// Decode a Pack in a new thread and send the CacheObjects while decoding.
    /// <br> Attention: It will consume the `pack` and return in a JoinHandle.
    pub fn decode_async(
        mut self,
        mut pack: impl BufRead + Send + 'static,
        sender: UnboundedSender<Entry>,
    ) -> JoinHandle<Pack> {
        let kind = get_hash_kind();
        thread::spawn(move || {
            set_hash_kind(kind);
            self.decode(
                &mut pack,
                move |entry| {
                    if let Err(e) = sender.send(entry.inner) {
                        eprintln!("Channel full, failed to send entry: {e:?}");
                    }
                },
                None::<fn(ObjectHash)>,
            )
            .unwrap();
            self
        })
    }

    /// Decodes a `Pack` from a `Stream` of `Bytes`, and sends the `Entry` while decoding.
    pub async fn decode_stream(
        mut self,
        mut stream: impl Stream<Item = Result<Bytes, Error>> + Unpin + Send + 'static,
        sender: UnboundedSender<MetaAttached<Entry, EntryMeta>>,
        pack_hash_send: Option<UnboundedSender<ObjectHash>>,
    ) -> Self {
        let kind = get_hash_kind();
        let (tx, rx) = std::sync::mpsc::channel();
        let mut reader = StreamBufReader::new(rx);
        tokio::spawn(async move {
            while let Some(chunk) = stream.next().await {
                let data = chunk.unwrap().to_vec();
                if let Err(e) = tx.send(data) {
                    eprintln!("Sending Error: {e:?}");
                    break;
                }
            }
        });
        // CPU-bound task, so use spawn_blocking
        // DO NOT use thread::spawn, because it will block tokio runtime (if single-threaded runtime, like in tests)
        tokio::task::spawn_blocking(move || {
            set_hash_kind(kind);
            self.decode(
                &mut reader,
                move |entry: MetaAttached<Entry, EntryMeta>| {
                    // as we used unbound channel here, it will never full so can be send with synchronous
                    if let Err(e) = sender.send(entry) {
                        eprintln!("unbound channel Sending Error: {e:?}");
                    }
                },
                Some(move |pack_id: ObjectHash| {
                    if let Some(pack_id_send) = pack_hash_send
                        && let Err(e) = pack_id_send.send(pack_id)
                    {
                        eprintln!("unbound channel Sending Error: {e:?}");
                    }
                }),
            )
            .unwrap();
            self
        })
        .await
        .unwrap()
    }

    /// CacheObjects + Index size of Caches
    fn memory_used(&self) -> usize {
        self.cache_objs_mem_used() + self.caches.memory_used_index()
    }

    /// The total memory used by the CacheObjects of this Pack
    fn cache_objs_mem_used(&self) -> usize {
        self.cache_objs_mem.load(Ordering::Acquire)
    }

    /// Rebuild the Delta Object in a new thread & process the objects waiting for it recursively.
    /// <br> This function must be *static*, because [&self] can't be moved into a new thread.
    fn process_delta(
        shared_params: Arc<SharedParams>,
        delta_obj: CacheObject,
        base_obj: Arc<CacheObject>,
    ) {
        shared_params.pool.clone().execute(move || {
            let mut new_obj = match delta_obj.info {
                CacheObjectInfo::OffsetDelta(_, _) | CacheObjectInfo::HashDelta(_, _) => {
                    Pack::rebuild_delta(delta_obj, base_obj)
                }
                CacheObjectInfo::OffsetZstdelta(_, _) => {
                    Pack::rebuild_zstdelta(delta_obj, base_obj)
                }
                _ => unreachable!(),
            };

            new_obj.set_mem_recorder(shared_params.cache_objs_mem_size.clone());
            new_obj.record_mem_size();
            Self::cache_obj_and_process_waitlist(shared_params, new_obj); //Indirect Recursion
        });
    }

    /// Cache the new object & process the objects waiting for it (in multi-threading).
    fn cache_obj_and_process_waitlist(shared_params: Arc<SharedParams>, new_obj: CacheObject) {
        (shared_params.callback)(new_obj.to_entry_metadata());
        let new_obj = shared_params.caches.insert(
            new_obj.offset,
            new_obj.base_object_hash().unwrap(),
            new_obj,
        );
        Self::process_waitlist(shared_params, new_obj);
    }

    fn process_waitlist(shared_params: Arc<SharedParams>, base_obj: Arc<CacheObject>) {
        let wait_objs = shared_params
            .waitlist
            .take(base_obj.offset, base_obj.base_object_hash().unwrap());
        for obj in wait_objs {
            // Process the objects waiting for the new object(base_obj = new_obj)
            Self::process_delta(shared_params.clone(), obj, base_obj.clone());
        }
    }

    /// Reconstruct the Delta Object based on the "base object"
    /// and return the new object.
    pub fn rebuild_delta(delta_obj: CacheObject, base_obj: Arc<CacheObject>) -> CacheObject {
        const COPY_INSTRUCTION_FLAG: u8 = 1 << 7;
        const COPY_OFFSET_BYTES: u8 = 4;
        const COPY_SIZE_BYTES: u8 = 3;
        const COPY_ZERO_SIZE: usize = 0x10000;

        let mut stream = Cursor::new(&delta_obj.data_decompressed);

        // Read the base object size
        // (Size Encoding)
        let (base_size, result_size) = utils::read_delta_object_size(&mut stream).unwrap();

        // Get the base object data
        let base_info = &base_obj.data_decompressed;
        assert_eq!(base_info.len(), base_size, "Base object size mismatch");

        let mut result = Vec::with_capacity(result_size);

        loop {
            // Check if the stream has ended, meaning the new object is done
            let instruction = match utils::read_bytes(&mut stream) {
                Ok([instruction]) => instruction,
                Err(err) if err.kind() == ErrorKind::UnexpectedEof => break,
                Err(err) => {
                    panic!(
                        "{}",
                        GitError::DeltaObjectError(format!("Wrong instruction in delta :{err}"))
                    );
                }
            };

            if instruction & COPY_INSTRUCTION_FLAG == 0 {
                // Data instruction; the instruction byte specifies the number of data bytes
                if instruction == 0 {
                    // Appending 0 bytes doesn't make sense, so git disallows it
                    panic!(
                        "{}",
                        GitError::DeltaObjectError(String::from("Invalid data instruction"))
                    );
                }

                // Append the provided bytes
                let mut data = vec![0; instruction as usize];
                stream.read_exact(&mut data).unwrap();
                result.extend_from_slice(&data);
            } else {
                // Copy instruction
                // +----------+---------+---------+---------+---------+-------+-------+-------+
                // | 1xxxxxxx | offset1 | offset2 | offset3 | offset4 | size1 | size2 | size3 |
                // +----------+---------+---------+---------+---------+-------+-------+-------+
                let mut nonzero_bytes = instruction;
                let offset =
                    utils::read_partial_int(&mut stream, COPY_OFFSET_BYTES, &mut nonzero_bytes)
                        .unwrap();
                let mut size =
                    utils::read_partial_int(&mut stream, COPY_SIZE_BYTES, &mut nonzero_bytes)
                        .unwrap();
                if size == 0 {
                    // Copying 0 bytes doesn't make sense, so git assumes a different size
                    size = COPY_ZERO_SIZE;
                }
                // Copy bytes from the base object
                let base_data = base_info.get(offset..(offset + size)).ok_or_else(|| {
                    GitError::DeltaObjectError("Invalid copy instruction".to_string())
                });

                match base_data {
                    Ok(data) => result.extend_from_slice(data),
                    Err(e) => panic!("{}", e),
                }
            }
        }
        assert_eq!(result_size, result.len(), "Result size mismatch");

        let hash = utils::calculate_object_hash(base_obj.object_type(), &result);
        // create new obj from `delta_obj` & `result` instead of modifying `delta_obj` for heap-size recording
        CacheObject {
            info: CacheObjectInfo::BaseObject(base_obj.object_type(), hash),
            offset: delta_obj.offset,
            crc32: delta_obj.crc32,
            data_decompressed: result,
            mem_recorder: None,
            is_delta_in_pack: delta_obj.is_delta_in_pack,
        } // Canonical form (Complete Object)
        // Memory recording will happen after this function returns. See `process_delta`
    }
    pub fn rebuild_zstdelta(delta_obj: CacheObject, base_obj: Arc<CacheObject>) -> CacheObject {
        let result = zstdelta::apply(&base_obj.data_decompressed, &delta_obj.data_decompressed)
            .expect("Failed to apply zstdelta");
        let hash = utils::calculate_object_hash(base_obj.object_type(), &result);
        CacheObject {
            info: CacheObjectInfo::BaseObject(base_obj.object_type(), hash),
            offset: delta_obj.offset,
            crc32: delta_obj.crc32,
            data_decompressed: result,
            mem_recorder: None,
            is_delta_in_pack: delta_obj.is_delta_in_pack,
        } // Canonical form (Complete Object)
        // Memory recording will happen after this function returns. See `process_delta`
    }
}

#[cfg(test)]
mod tests {
    use std::{
        env, fs,
        io::{BufReader, Cursor, prelude::*},
        path::PathBuf,
        sync::{
            Arc,
            atomic::{AtomicUsize, Ordering},
        },
    };

    use flate2::{Compression, write::ZlibEncoder};
    use futures_util::TryStreamExt;
    use tokio_util::io::ReaderStream;

    use crate::{
        hash::{HashKind, ObjectHash, set_hash_kind},
        internal::pack::{Pack, tests::init_logger},
    };

    #[tokio::test]
    async fn test_pack_check_header() {
        let mut source = PathBuf::from(env!("CARGO_MANIFEST_DIR"));
        source.push("tests/data/packs/medium-sha1.pack");

        let f = fs::File::open(source).unwrap();
        let mut buf_reader = BufReader::new(f);
        let (object_num, _) = Pack::check_header(&mut buf_reader).unwrap();

        assert_eq!(object_num, 35031);
    }

    #[test]
    fn test_decompress_data() {
        let data = b"Hello, world!"; // Sample data to compress and then decompress
        let mut encoder = ZlibEncoder::new(Vec::new(), Compression::default());
        encoder.write_all(data).unwrap();
        let compressed_data = encoder.finish().unwrap();
        let compressed_size = compressed_data.len();

        // Create a cursor for the compressed data to simulate a BufRead source
        let mut cursor: Cursor<Vec<u8>> = Cursor::new(compressed_data);
        let expected_size = data.len();

        // Decompress the data and assert correctness
        let result = Pack::decompress_data(&mut cursor, expected_size);
        match result {
            Ok((decompressed_data, bytes_read)) => {
                assert_eq!(bytes_read, compressed_size);
                assert_eq!(decompressed_data, data);
            }
            Err(e) => panic!("Decompression failed: {e:?}"),
        }
    }

    /// Helper function to run decode tests without delta objects
    fn run_decode_no_delta(rel_path: &str, kind: HashKind) {
        let _guard = set_hash_kind(kind);
        let mut source = PathBuf::from(env!("CARGO_MANIFEST_DIR"));
        source.push(rel_path);

        let tmp = PathBuf::from("/tmp/.cache_temp");

        let f = fs::File::open(source).unwrap();
        let mut buffered = BufReader::new(f);
        let mut p = Pack::new(None, Some(1024 * 1024 * 20), Some(tmp), true);
        p.decode(&mut buffered, |_| {}, None::<fn(ObjectHash)>)
            .unwrap();
    }
    #[test]
    fn test_pack_decode_without_delta() {
        run_decode_no_delta("tests/data/packs/small-sha1.pack", HashKind::Sha1);
        run_decode_no_delta("tests/data/packs/small-sha256.pack", HashKind::Sha256);
    }

    /// Helper function to run decode tests with delta objects
    fn run_decode_with_ref_delta(rel_path: &str, kind: HashKind) {
        let _guard = set_hash_kind(kind);
        init_logger();

        let mut source = PathBuf::from(env!("CARGO_MANIFEST_DIR"));
        source.push(rel_path);

        let tmp = PathBuf::from("/tmp/.cache_temp");

        let f = fs::File::open(source).unwrap();
        let mut buffered = BufReader::new(f);
        let mut p = Pack::new(None, Some(1024 * 1024 * 20), Some(tmp), true);
        p.decode(&mut buffered, |_| {}, None::<fn(ObjectHash)>)
            .unwrap();
    }
    #[test]
    fn test_pack_decode_with_ref_delta() {
        run_decode_with_ref_delta("tests/data/packs/ref-delta-sha1.pack", HashKind::Sha1);
        run_decode_with_ref_delta("tests/data/packs/ref-delta-sha256.pack", HashKind::Sha256);
    }

    /// Helper function to run decode tests without memory limit
    fn run_decode_no_mem_limit(rel_path: &str, kind: HashKind) {
        let _guard = set_hash_kind(kind);
        let mut source = PathBuf::from(env!("CARGO_MANIFEST_DIR"));
        source.push(rel_path);

        let tmp = PathBuf::from("/tmp/.cache_temp");

        let f = fs::File::open(source).unwrap();
        let mut buffered = BufReader::new(f);
        let mut p = Pack::new(None, None, Some(tmp), true);
        p.decode(&mut buffered, |_| {}, None::<fn(ObjectHash)>)
            .unwrap();
    }
    #[test]
    fn test_pack_decode_no_mem_limit() {
        run_decode_no_mem_limit("tests/data/packs/small-sha1.pack", HashKind::Sha1);
        run_decode_no_mem_limit("tests/data/packs/small-sha256.pack", HashKind::Sha256);
    }

    /// Helper function to run decode tests with delta objects
    async fn run_decode_large_with_delta(rel_path: &str, kind: HashKind) {
        let _guard = set_hash_kind(kind);
        init_logger();
        let mut source = PathBuf::from(env!("CARGO_MANIFEST_DIR"));
        source.push(rel_path);

        let tmp = PathBuf::from("/tmp/.cache_temp");

        let f = fs::File::open(source).unwrap();
        let mut buffered = BufReader::new(f);
        let mut p = Pack::new(
            Some(4),
            Some(1024 * 1024 * 100), //try to avoid dead lock on CI servers with low memory
            Some(tmp.clone()),
            true,
        );
        let rt = p.decode(
            &mut buffered,
            |_obj| {
                // println!("{:?} {}", obj.hash.to_string(), offset);
            },
            None::<fn(ObjectHash)>,
        );
        if let Err(e) = rt {
            fs::remove_dir_all(tmp).unwrap();
            panic!("Error: {e:?}");
        }
    }
    #[tokio::test]
    async fn test_pack_decode_with_large_file_with_delta_without_ref() {
        run_decode_large_with_delta("tests/data/packs/medium-sha1.pack", HashKind::Sha1).await;
        run_decode_large_with_delta("tests/data/packs/medium-sha256.pack", HashKind::Sha256).await;
    } // it will be stuck on dropping `Pack` on Windows if `mem_size` is None, so we need `mimalloc`

    /// Helper function to run decode tests with large file stream
    async fn run_decode_large_stream(rel_path: &str, kind: HashKind) {
        let _guard = set_hash_kind(kind);
        init_logger();
        let mut source = PathBuf::from(env!("CARGO_MANIFEST_DIR"));
        source.push(rel_path);

        let tmp = PathBuf::from("/tmp/.cache_temp");
        let f = tokio::fs::File::open(source).await.unwrap();
        let stream = ReaderStream::new(f).map_err(axum::Error::new);
        let p = Pack::new(Some(4), Some(1024 * 1024 * 100), Some(tmp.clone()), true);

        let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel();
        let handle = tokio::spawn(async move { p.decode_stream(stream, tx, None).await });
        let count = Arc::new(AtomicUsize::new(0));
        let count_c = count.clone();
        // in tests, RUNTIME is single-threaded, so `sync code` will block the tokio runtime
        let consume = tokio::spawn(async move {
            let mut cnt = 0;
            while let Some(_entry) = rx.recv().await {
                cnt += 1;
            }
            tracing::info!("Received: {}", cnt);
            count_c.store(cnt, Ordering::Release);
        });
        let p = handle.await.unwrap();
        consume.await.unwrap();
        assert_eq!(count.load(Ordering::Acquire), p.number);
        assert_eq!(p.number, 35031);
    }
    #[tokio::test]
    async fn test_decode_large_file_stream() {
        run_decode_large_stream("tests/data/packs/medium-sha1.pack", HashKind::Sha1).await;
        run_decode_large_stream("tests/data/packs/medium-sha256.pack", HashKind::Sha256).await;
    }

    /// Helper function to run decode tests with large file async
    async fn run_decode_large_file_async(rel_path: &str, kind: HashKind) {
        let _guard = set_hash_kind(kind);
        let mut source = PathBuf::from(env!("CARGO_MANIFEST_DIR"));
        source.push(rel_path);

        let tmp = PathBuf::from("/tmp/.cache_temp");
        let f = fs::File::open(source).unwrap();
        let buffered = BufReader::new(f);
        let p = Pack::new(Some(4), Some(1024 * 1024 * 100), Some(tmp.clone()), true);

        let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel();
        let handle = p.decode_async(buffered, tx); // new thread
        let mut cnt = 0;
        while let Some(_entry) = rx.recv().await {
            cnt += 1; //use entry here
        }
        let p = handle.join().unwrap();
        assert_eq!(cnt, p.number);
    }
    #[tokio::test]
    async fn test_decode_large_file_async() {
        run_decode_large_file_async("tests/data/packs/medium-sha1.pack", HashKind::Sha1).await;
        run_decode_large_file_async("tests/data/packs/medium-sha256.pack", HashKind::Sha256).await;
    }

    /// Helper function to run decode tests with delta objects without reference
    fn run_decode_with_delta_no_ref(rel_path: &str, kind: HashKind) {
        let _guard = set_hash_kind(kind);
        let mut source = PathBuf::from(env!("CARGO_MANIFEST_DIR"));
        source.push(rel_path);

        let tmp = PathBuf::from("/tmp/.cache_temp");

        let f = fs::File::open(source).unwrap();
        let mut buffered = BufReader::new(f);
        let mut p = Pack::new(None, Some(1024 * 1024 * 20), Some(tmp), true);
        p.decode(&mut buffered, |_| {}, None::<fn(ObjectHash)>)
            .unwrap();
    }
    #[test]
    fn test_pack_decode_with_delta_without_ref() {
        run_decode_with_delta_no_ref("tests/data/packs/medium-sha1.pack", HashKind::Sha1);
        run_decode_with_delta_no_ref("tests/data/packs/medium-sha256.pack", HashKind::Sha256);
    }

    #[test] // Take too long time
    fn test_pack_decode_multi_task_with_large_file_with_delta_without_ref() {
        let rt = tokio::runtime::Builder::new_current_thread()
            .enable_all()
            .build()
            .unwrap();
        rt.block_on(async move {
            // For each hash kind, run two decode tasks concurrently to simulate multi-task pressure.
            for (kind, path) in [
                (HashKind::Sha1, "tests/data/packs/medium-sha1.pack"),
                (HashKind::Sha256, "tests/data/packs/medium-sha256.pack"),
            ] {
                let f1 = run_decode_large_with_delta(path, kind);
                let f2 = run_decode_large_with_delta(path, kind);
                let _ = futures::future::join(f1, f2).await;
            }
        });
    }
}