gix-pack 0.76.0

use gix_error::Result;
use std::sync::atomic::{AtomicBool, Ordering};

use gix_error::bail;
use gix_parallel as threading;
use gix_parallel::{Mutable, OwnShared};
use gix_utils::progress::{self, DynNestedProgress, Progress};

use crate::{
    cache::delta::{Tree, traverse::util::ItemSliceSync, tree::Item},
    data::EntryRange,
};

mod resolve;
pub(crate) mod util;

/// Shared access to ref-delta child indices awaiting a resolved base, keyed by its object ID.
pub(super) type SharedRefDeltaChildren = OwnShared<Mutable<super::tree::RefDeltaChildren>>;

#[cold]
pub(super) fn allocation_error(kind: gix_error::ResourceExhaustionKind) -> gix_error::Message {
    gix_error::resource_exhaustion(kind, "Entry too large to fit in memory")
}

pub(super) fn interrupted() -> gix_error::Error {
    gix_error::message("Interrupted").cancelled_error()
}

/// Additional context passed to the `inspect_object(…)` function of the [`Tree::traverse()`] method.
pub struct Context<'a> {
    /// The pack entry describing the object
    pub entry: &'a crate::data::Entry,
    /// The offset at which `entry` ends in the pack, useful to learn about the exact range of `entry` within the pack.
    pub entry_end: u64,
    /// The decompressed object itself, ready to be decoded.
    pub decompressed: &'a [u8],
    /// The depth at which this object resides in the delta-tree. It represents the number of base objects, with 0 indicating
    /// an 'undeltified' object, and higher values indicating delta objects with the given number of bases.
    pub level: u16,
}

/// Options for [`Tree::traverse()`].
pub struct Options<'a, 's> {
    /// is a progress instance to track progress for each object in the traversal.
    pub object_progress: Box<dyn DynNestedProgress>,
    /// is a progress instance to track the overall progress.
    pub size_progress: &'s mut dyn Progress,
    /// If `Some`, only use the given number of threads. Otherwise, the number of threads to use will be selected based on
    /// the number of available logical cores.
    pub thread_limit: Option<usize>,
    /// Abort the operation if the value is `true`.
    pub should_interrupt: &'a AtomicBool,
    /// specifies what kind of hashes we expect to be stored in oid-delta entries, which is viable to decoding them
    /// with the correct size.
    pub object_hash: gix_hash::Kind,
    /// If `Some`, rejects individual allocations above the given number of bytes while resolving decoded object and
    /// delta result buffers. `Some(0)` rejects all non-empty allocations.
    pub alloc_limit_bytes: Option<usize>,
}

/// The outcome of [`Tree::traverse()`]
pub struct Outcome<T> {
    /// The items that have no children in the pack, i.e. base objects.
    pub roots: Vec<Item<T>>,
    /// The items that children to a root object, i.e. delta objects.
    pub children: Vec<Item<T>>,
}

impl<T> Tree<T>
where
    T: Send,
{
    /// Traverse this tree of delta objects with a function `inspect_object` to process each object at will.
    ///
    /// Forward references with known base offsets are attached before traversal. Children recorded by base object ID
    /// during streaming indexing are attached when their bases are fully decoded and hashed, including bases that are
    /// themselves deltas. All bases must be represented in this tree; external thin-pack bases must be inserted beforehand.
    ///
    /// * `should_run_in_parallel() -> bool` returns true if the underlying pack is big enough to warrant parallel traversal at all.
    /// * `resolve(EntrySlice, &mut Vec<u8>) -> Option<()>` resolves the bytes in the pack for the given `EntrySlice` and stores them in the
    ///   output vector. It returns `Some(())` if the object existed in the pack, or `None` to indicate a resolution error, which would abort the
    ///   operation as well.
    /// * `pack_entries_end` marks one-past-the-last byte of the last entry in the pack, as the last entries size would otherwise
    ///   be unknown as it's not part of the index file.
    /// * `inspect_object(node_data: &mut T, progress: Progress, context: Context<ThreadLocal State>) -> Result<(), CustomError>` is a function
    ///   running for each thread receiving fully decoded objects along with contextual information, which either succeeds with `Ok(())`
    ///   or returns a `CustomError`.
    ///   Note that `node_data` can be modified to allow storing maintaining computation results on a per-object basis. It should contain
    ///   its own mutable per-thread data as required.
    ///
    /// This method returns a vector of all tree items, along with their potentially modified custom node data.
    ///
    /// _Note_ that this method consumed the Tree to assure safe parallel traversal with mutation support.
    pub fn traverse<F, MBFN, R>(
        mut self,
        resolve: F,
        resolve_data: &R,
        pack_entries_end: u64,
        inspect_object: MBFN,
        Options {
            thread_limit,
            mut object_progress,
            size_progress,
            should_interrupt,
            object_hash,
            alloc_limit_bytes,
        }: Options<'_, '_>,
    ) -> Result<Outcome<T>>
    where
        F: for<'r> Fn(EntryRange, &'r R) -> Option<&'r [u8]> + Send + Clone,
        R: Send + Sync,
        MBFN: FnMut(&mut T, &dyn Progress, Context<'_>) -> Result + Send + Clone,
    {
        self.set_pack_entries_end_and_resolve_ref_offsets(pack_entries_end)?;

        let num_objects = self.num_items();
        let object_counter = {
            let progress = &mut object_progress;
            progress.init(Some(num_objects), progress::count("objects"));
            progress.set(0);
            progress.counter()
        };
        size_progress.init(None, progress::bytes());
        let size_counter = size_progress.counter();
        let resolver_progress = object_progress.add_child("delta resolver".into());

        let start = std::time::Instant::now();
        let (mut root_items, mut child_items_vec, ref_delta_children) = self.take_root_child_and_refs();
        let ref_delta_children =
            (!ref_delta_children.is_empty()).then(|| OwnShared::new(Mutable::new(ref_delta_children)));
        let child_items = ItemSliceSync::new(&mut child_items_vec);
        // SAFETY: Both item slices come from the same Tree, whose child-index uniqueness invariant still holds.
        #[expect(unsafe_code)]
        unsafe {
            resolve::all(
                &mut root_items,
                &child_items,
                thread_limit,
                num_objects,
                object_counter.clone(),
                size_counter,
                &resolver_progress,
                resolve,
                resolve_data,
                inspect_object,
                ref_delta_children.clone(),
                object_hash,
                alloc_limit_bytes,
                should_interrupt,
            )?;
        }

        if let Some(ref_delta_children) = ref_delta_children
            && let Some((base_id, _children)) = threading::lock(&ref_delta_children).first_key_value()
        {
            bail!("The ref-delta base object {base_id} could not be found".not_found());
        }

        if object_counter.load(Ordering::Relaxed) != num_objects {
            bail!(gix_error::corruption("Pack delta traversal left unresolved objects"));
        }

        object_progress.show_throughput(start);
        size_progress.show_throughput(start);

        Ok(Outcome {
            roots: root_items,
            children: child_items_vec,
        })
    }
}