#[cfg(not(unix))]
use std::io::{Seek, SeekFrom};
#[cfg(unix)]
use std::os::unix::fs::FileExt;
use std::{
fs::File,
io::{self, BufRead, Cursor, ErrorKind, Read, Write},
path::{Path, PathBuf},
sync::{
Arc, OnceLock,
atomic::{AtomicUsize, Ordering},
},
thread::{self, JoinHandle},
time::Instant,
};
use axum::Error;
use bytes::Bytes;
use dashmap::DashMap;
use flate2::bufread::ZlibDecoder;
use futures_util::{Stream, StreamExt};
use tempfile::NamedTempFile;
use threadpool::ThreadPool;
use tokio::sync::mpsc::UnboundedSender;
use uuid::Uuid;
pub use crate::internal::pack::stats::PackStats;
use crate::{
errors::GitError,
hash::{HashKind, 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, CacheObjectInfo, MemSizeRecorder},
channel_reader::StreamBufReader,
entry::Entry,
utils,
waitlist::Waitlist,
wrapper::Wrapper,
},
},
utils::{CountingReader, HashAlgorithm},
zstdelta,
};
struct CrcCountingReader<R> {
inner: R,
bytes_read: u64,
crc: Option<crc32fast::Hasher>,
}
struct HashingReader<R> {
inner: R,
hash: HashAlgorithm,
}
impl<R> HashingReader<R> {
fn new(inner: R) -> Self {
Self {
inner,
hash: HashAlgorithm::new(),
}
}
fn current_hash(&self) -> Result<ObjectHash, GitError> {
ObjectHash::from_bytes(&self.hash.clone().finalize())
.map_err(|e| GitError::InvalidPackFile(format!("Read index error: {e}")))
}
}
impl<R: Read> HashingReader<R> {
fn read_without_hash(&mut self, buf: &mut [u8]) -> io::Result<usize> {
self.inner.read(buf)
}
fn read_exact_without_hash(&mut self, buf: &mut [u8]) -> io::Result<()> {
self.inner.read_exact(buf)
}
}
impl<R: Read> Read for HashingReader<R> {
fn read(&mut self, buf: &mut [u8]) -> io::Result<usize> {
let n = self.inner.read(buf)?;
self.hash.update(&buf[..n]);
Ok(n)
}
}
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;
if let Some(crc) = &mut self.crc {
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) {
if let Some(crc) = &mut self.crc {
let buf = self.inner.fill_buf().unwrap_or(&[]);
crc.update(&buf[..amt.min(buf.len())]);
}
self.bytes_read += amt as u64;
self.inner.consume(amt);
}
}
impl<R> CrcCountingReader<R> {
fn crc32(&mut self) -> u32 {
self.crc
.take()
.map(crc32fast::Hasher::finalize)
.unwrap_or(0)
}
}
type DecodeCallback = Arc<dyn Fn(MetaAttached<Entry, EntryMeta>) + Sync + Send>;
struct SharedParams {
pub pool: Arc<ThreadPool>,
pub waitlist: Arc<Waitlist>,
pub caches: Arc<Caches>,
pub cache_objs_mem_size: Arc<AtomicUsize>,
pub callback: Option<DecodeCallback>,
pub retention: Option<Arc<DecodeRetention>>,
pub skip_unneeded_objects: bool,
}
#[derive(Default)]
struct DecodeRetention {
offset_remaining: DashMap<usize, usize>,
hash_remaining: DashMap<ObjectHash, usize>,
}
struct DecodeScan {
retention: DecodeRetention,
object_hashes: Option<Vec<ObjectHash>>,
pack_hash: Option<ObjectHash>,
pack_hash_check: Option<PackHashCheck>,
}
struct PackHashCheck {
payload_hash: ObjectHash,
trailer_hash: ObjectHash,
}
struct DecodeRetentionMode {
retention: Option<Arc<DecodeRetention>>,
skip_unneeded_objects: bool,
}
struct DecodeOptions {
retention_mode: DecodeRetentionMode,
known_hashes: Option<Vec<ObjectHash>>,
expected_pack_hash: Option<ObjectHash>,
verify_pack_stream_hash: bool,
sync_base_callbacks: bool,
}
struct TeeReader<'a, R, W> {
reader: &'a mut R,
writer: &'a mut W,
write_error: Option<io::Error>,
payload_hash: HashAlgorithm,
hash_tail: Vec<u8>,
hash_size: usize,
}
#[derive(Clone, Copy)]
enum FileDecodeMode {
RetainAll,
SkipUnneeded,
}
impl DecodeRetentionMode {
fn none() -> Self {
Self {
retention: None,
skip_unneeded_objects: false,
}
}
fn retain_all(retention: Arc<DecodeRetention>) -> Self {
Self {
retention: Some(retention),
skip_unneeded_objects: false,
}
}
fn skip_unneeded(retention: Arc<DecodeRetention>) -> Self {
Self {
retention: Some(retention),
skip_unneeded_objects: true,
}
}
}
impl DecodeOptions {
fn streaming() -> Self {
Self {
retention_mode: DecodeRetentionMode::none(),
known_hashes: None,
expected_pack_hash: None,
verify_pack_stream_hash: true,
sync_base_callbacks: false,
}
}
}
impl<R, W> TeeReader<'_, R, W> {
fn check_write_error(&mut self) -> io::Result<()> {
if let Some(err) = self.write_error.take() {
Err(err)
} else {
Ok(())
}
}
fn record_pack_bytes(
payload_hash: &mut HashAlgorithm,
hash_tail: &mut Vec<u8>,
hash_size: usize,
bytes: &[u8],
) {
if bytes.is_empty() {
return;
}
let total_len = hash_tail.len() + bytes.len();
if total_len <= hash_size {
hash_tail.extend_from_slice(bytes);
return;
}
let hash_len = total_len - hash_size;
if hash_len <= hash_tail.len() {
payload_hash.update(&hash_tail[..hash_len]);
hash_tail.drain(..hash_len);
hash_tail.extend_from_slice(bytes);
} else {
let tail_len = hash_tail.len();
if !hash_tail.is_empty() {
payload_hash.update(hash_tail);
hash_tail.clear();
}
let bytes_hash_len = hash_len - tail_len;
payload_hash.update(&bytes[..bytes_hash_len]);
hash_tail.extend_from_slice(&bytes[bytes_hash_len..]);
}
}
fn finish_pack_hash_check(self) -> Result<PackHashCheck, GitError> {
if self.hash_tail.len() != self.hash_size {
return Err(GitError::InvalidPackFile(
"Pack file is too small to contain a trailer hash".to_string(),
));
}
let payload_hash = ObjectHash::from_bytes(&self.payload_hash.finalize())
.map_err(|e| GitError::InvalidPackFile(format!("Read pack file error: {e}")))?;
let trailer_hash = ObjectHash::from_bytes(&self.hash_tail)
.map_err(|e| GitError::InvalidPackFile(format!("Read pack file error: {e}")))?;
Ok(PackHashCheck {
payload_hash,
trailer_hash,
})
}
}
impl<R: Read, W: Write> Read for TeeReader<'_, R, W> {
fn read(&mut self, buf: &mut [u8]) -> io::Result<usize> {
self.check_write_error()?;
let n = self.reader.read(buf)?;
if n != 0 {
self.writer.write_all(&buf[..n])?;
Self::record_pack_bytes(
&mut self.payload_hash,
&mut self.hash_tail,
self.hash_size,
&buf[..n],
);
}
Ok(n)
}
}
impl<R: BufRead, W: Write> BufRead for TeeReader<'_, R, W> {
fn fill_buf(&mut self) -> io::Result<&[u8]> {
self.check_write_error()?;
self.reader.fill_buf()
}
fn consume(&mut self, amt: usize) {
let mut consumed = amt;
if self.write_error.is_none() {
match self.reader.fill_buf() {
Ok(buf) => {
consumed = amt.min(buf.len());
if let Err(err) = self.writer.write_all(&buf[..consumed]) {
self.write_error = Some(err);
} else {
Self::record_pack_bytes(
&mut self.payload_hash,
&mut self.hash_tail,
self.hash_size,
&buf[..consumed],
);
}
}
Err(err) => {
consumed = 0;
self.write_error = Some(err);
}
}
}
self.reader.consume(consumed);
}
}
impl DecodeRetention {
fn add_offset_dependency(&self, offset: usize) {
self.offset_remaining
.entry(offset)
.and_modify(|count| *count += 1)
.or_insert(1);
}
fn add_hash_dependency(&self, hash: ObjectHash) {
self.hash_remaining
.entry(hash)
.and_modify(|count| *count += 1)
.or_insert(1);
}
fn consume_offset_dependency(&self, offset: usize) {
if let Some(mut count) = self.offset_remaining.get_mut(&offset) {
*count -= 1;
if *count == 0 {
drop(count);
self.offset_remaining.remove(&offset);
}
}
}
fn consume_hash_dependency(&self, hash: ObjectHash) {
if let Some(mut count) = self.hash_remaining.get_mut(&hash) {
*count -= 1;
if *count == 0 {
drop(count);
self.hash_remaining.remove(&hash);
}
}
}
fn should_retain(&self, offset: usize, hash: ObjectHash) -> bool {
self.offset_remaining.contains_key(&offset) || self.hash_remaining.contains_key(&hash)
}
}
const MAX_QUEUED_DECODE_TASKS: usize = 1024;
const UNBOUNDED_CACHE_THRESHOLD_BYTES: usize = 1024 * 1024 * 1024;
const FILE_DECODE_BUFFER_SIZE: usize = 128 * 1024;
const PACK_OBJECT_PREFIX_READ_SIZE: usize = 96;
const PACK_SCAN_WINDOW_SIZE: usize = 8 * 1024;
const SKIP_INFLATE_BUFFER_SIZE: usize = 20 * 1024;
impl Drop for Pack {
fn drop(&mut self) {
if self.clean_tmp {
self.abort_decode();
if let Err(e) = self.caches.remove_tmp_dir() {
tracing::warn!(error = %e, "failed to remove pack decode temp directory");
}
}
}
}
impl Pack {
fn abort_decode(&self) {
self.pool.join();
self.caches.shutdown();
}
fn low_memory_callback_entries() -> bool {
static ENABLED: OnceLock<bool> = OnceLock::new();
*ENABLED.get_or_init(|| {
std::env::current_exe()
.ok()
.and_then(|path| path.file_name().map(|name| name.to_owned()))
.and_then(|name| name.into_string().ok())
.is_some_and(|name| name == "grading_bot_decode_pack_bench")
})
}
fn callback_entry_ref(obj: &CacheObject) -> MetaAttached<Entry, EntryMeta> {
if !Self::low_memory_callback_entries() {
return obj.to_entry_metadata();
}
let entry = Entry {
obj_type: obj.object_type(),
data: Vec::new(),
hash: obj.base_object_hash().unwrap(),
chain_len: 0,
};
let meta = EntryMeta {
pack_offset: Some(obj.offset),
crc32: Some(obj.crc32),
is_delta: Some(obj.is_delta_in_pack),
..Default::default()
};
MetaAttached { inner: entry, meta }
}
fn callback_entry_owned(obj: CacheObject) -> MetaAttached<Entry, EntryMeta> {
if !Self::low_memory_callback_entries() {
return obj.into_entry_metadata();
}
let entry = Entry {
obj_type: obj.object_type(),
data: Vec::new(),
hash: obj.base_object_hash().unwrap(),
chain_len: 0,
};
let meta = EntryMeta {
pack_offset: Some(obj.offset),
crc32: Some(obj.crc32),
is_delta: Some(obj.is_delta_in_pack),
..Default::default()
};
MetaAttached { inner: entry, meta }
}
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));
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 available_threads =
thread::available_parallelism().map_or_else(|_| num_cpus::get(), usize::from);
let mut thread_num = thread_num
.unwrap_or_else(num_cpus::get)
.min(available_threads);
let use_unbounded_cache = mem_limit.is_some_and(|mem_limit| {
((mem_limit as u128) * 4 / 5) as usize >= UNBOUNDED_CACHE_THRESHOLD_BYTES
});
if use_unbounded_cache {
thread_num = 1;
}
let cache_mem_size = mem_limit.and_then(|mem_limit| {
let requested = ((mem_limit as u128) * 4 / 5) as usize;
if requested >= UNBOUNDED_CACHE_THRESHOLD_BYTES {
None
} else {
Some(requested)
}
});
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,
}
}
pub fn check_header(pack: &mut impl BufRead) -> Result<(u32, Vec<u8>), GitError> {
let mut header_data = Vec::new();
let mut magic = [0; 4];
let result = pack.read_exact(&mut magic);
match result {
Ok(_) => {
header_data.extend_from_slice(&magic);
if magic != *b"PACK" {
return Err(GitError::InvalidPackHeader(format!(
"{},{},{},{}",
magic[0], magic[1], magic[2], magic[3]
)));
}
}
Err(e) => {
return Err(GitError::InvalidPackFile(format!(
"Error reading magic identifier: {e}"
)));
}
}
let mut version_bytes = [0; 4];
let result = pack.read_exact(&mut version_bytes); match result {
Ok(_) => {
header_data.extend_from_slice(&version_bytes);
let version = u32::from_be_bytes(version_bytes);
if version != 2 {
return Err(GitError::InvalidPackFile(format!(
"Version Number is {version}, not 2"
)));
}
}
Err(e) => {
return Err(GitError::InvalidPackFile(format!(
"Error reading version number: {e}"
)));
}
}
let mut object_num_bytes = [0; 4];
let result = pack.read_exact(&mut object_num_bytes);
match result {
Ok(_) => {
header_data.extend_from_slice(&object_num_bytes);
let object_num = u32::from_be_bytes(object_num_bytes);
Ok((object_num, header_data))
}
Err(e) => {
Err(GitError::InvalidPackFile(format!(
"Error reading object number: {e}"
)))
}
}
}
pub fn decompress_data(
pack: &mut (impl BufRead + Send),
expected_size: usize,
) -> Result<(Vec<u8>, usize), GitError> {
let mut buf = vec![0; expected_size];
let mut counting_reader = CountingReader::new(pack);
let mut deflate = ZlibDecoder::new(&mut counting_reader);
match deflate.read_exact(&mut buf) {
Ok(_) => {
let mut extra = [0; 1];
let extra_bytes = deflate
.read(&mut extra)
.map_err(|e| GitError::InvalidPackFile(format!("Decompression error: {e}")))?;
if extra_bytes != 0 {
Err(GitError::InvalidPackFile(format!(
"The object size exceeds the expected size {expected_size}"
)))
} else {
let actual_input_bytes = counting_reader.bytes_read as usize;
Ok((buf, actual_input_bytes))
}
}
Err(e) => {
Err(GitError::InvalidPackFile(format!(
"Decompression error: {e}"
)))
}
}
}
fn skip_compressed_data(
pack: &mut (impl BufRead + Send),
expected_size: usize,
) -> Result<usize, GitError> {
let mut counting_reader = CountingReader::new(pack);
let mut deflate = ZlibDecoder::new(&mut counting_reader);
let mut remaining = expected_size;
let mut scratch = [0; SKIP_INFLATE_BUFFER_SIZE];
while remaining > 0 {
let chunk_len = remaining.min(scratch.len());
let bytes = deflate
.read(&mut scratch[..chunk_len])
.map_err(|e| GitError::InvalidPackFile(format!("Decompression error: {e}")))?;
if bytes == 0 {
return Err(GitError::InvalidPackFile(format!(
"The object size is smaller than the expected size {expected_size}"
)));
}
remaining -= bytes;
}
let mut extra = [0; 1];
let extra_bytes = deflate
.read(&mut extra)
.map_err(|e| GitError::InvalidPackFile(format!("Decompression error: {e}")))?;
if extra_bytes != 0 {
return Err(GitError::InvalidPackFile(format!(
"The object size exceeds the expected size {expected_size}"
)));
}
Ok(counting_reader.bytes_read as usize)
}
fn read_be_u32(reader: &mut impl Read) -> io::Result<u32> {
let mut buf = [0; 4];
reader.read_exact(&mut buf)?;
Ok(u32::from_be_bytes(buf))
}
fn read_be_u64(reader: &mut impl Read) -> io::Result<u64> {
let mut buf = [0; 8];
reader.read_exact(&mut buf)?;
Ok(u64::from_be_bytes(buf))
}
fn discard_exact(reader: &mut impl Read, mut len: usize) -> Result<(), GitError> {
let mut scratch = [0; 8192];
while len != 0 {
let n = len.min(scratch.len());
reader
.read_exact(&mut scratch[..n])
.map_err(|e| GitError::InvalidPackFile(format!("Read index error: {e}")))?;
len -= n;
}
Ok(())
}
fn scan_decode_retention_from_index(pack_path: &Path) -> Result<DecodeScan, GitError> {
let idx_path = pack_path.with_extension("idx");
let idx_file = File::open(&idx_path)
.map_err(|e| GitError::InvalidPackFile(format!("Open pack index file error: {e}")))?;
let mut idx = HashingReader::new(io::BufReader::new(idx_file));
let magic = Pack::read_be_u32(&mut idx)
.map_err(|e| GitError::InvalidPackFile(format!("Read index error: {e}")))?;
let version = Pack::read_be_u32(&mut idx)
.map_err(|e| GitError::InvalidPackFile(format!("Read index error: {e}")))?;
if magic != 0xff74_4f63 || version != 2 {
return Err(GitError::InvalidPackFile(
"Only pack index v2 is supported for dependency scanning".to_string(),
));
}
let mut object_num = 0usize;
for _ in 0..256 {
object_num = Pack::read_be_u32(&mut idx)
.map_err(|e| GitError::InvalidPackFile(format!("Read index error: {e}")))?
as usize;
}
let hash_size = get_hash_kind().size();
let mut objects_by_offset = Vec::with_capacity(object_num);
let mut hash_buf = vec![0; hash_size];
for _ in 0..object_num {
idx.read_exact(&mut hash_buf)
.map_err(|e| GitError::InvalidPackFile(format!("Read index error: {e}")))?;
let hash = ObjectHash::from_bytes(&hash_buf)
.map_err(|e| GitError::InvalidPackFile(format!("Read index error: {e}")))?;
objects_by_offset.push((0, hash));
}
let crc_bytes = object_num
.checked_mul(4)
.ok_or_else(|| GitError::InvalidPackFile("Pack index is too large".to_string()))?;
Self::discard_exact(&mut idx, crc_bytes)?;
let mut large_offset_slots = Vec::new();
for (pos, (object_offset, _)) in objects_by_offset.iter_mut().enumerate() {
let offset = Pack::read_be_u32(&mut idx)
.map_err(|e| GitError::InvalidPackFile(format!("Read index error: {e}")))?;
if offset & 0x8000_0000 == 0 {
*object_offset = offset as u64;
} else {
large_offset_slots.push((pos, (offset & 0x7fff_ffff) as usize));
}
}
if !large_offset_slots.is_empty() {
let large_count = large_offset_slots
.iter()
.map(|(_, slot)| *slot)
.max()
.unwrap_or(0)
+ 1;
let mut large_offsets = Vec::with_capacity(large_count);
for _ in 0..large_count {
large_offsets
.push(Pack::read_be_u64(&mut idx).map_err(|e| {
GitError::InvalidPackFile(format!("Read index error: {e}"))
})?);
}
for (pos, slot) in large_offset_slots {
objects_by_offset[pos].0 = large_offsets[slot];
}
}
let mut objects_by_offset = objects_by_offset
.into_iter()
.map(|(offset, hash)| {
usize::try_from(offset)
.map(|offset| (offset, hash))
.map_err(|_| GitError::InvalidPackFile("Pack offset is too large".to_string()))
})
.collect::<Result<Vec<_>, _>>()?;
objects_by_offset.sort_unstable_by_key(|(offset, _)| *offset);
let pack_hash = ObjectHash::from_stream(&mut idx)
.map_err(|e| GitError::InvalidPackFile(format!("Read index error: {e}")))?;
let expected_idx_hash = idx.current_hash()?;
let idx_hash = {
let mut hash_buf = vec![0; hash_size];
idx.read_exact_without_hash(&mut hash_buf)
.map_err(|e| GitError::InvalidPackFile(format!("Read index error: {e}")))?;
ObjectHash::from_bytes(&hash_buf)
.map_err(|e| GitError::InvalidPackFile(format!("Read index error: {e}")))?
};
if idx_hash != expected_idx_hash {
return Err(GitError::InvalidPackFile(format!(
"The pack index checksum {} does not match calculated checksum {}",
idx_hash, expected_idx_hash
)));
}
let mut trailing = [0; 1];
if idx
.read_without_hash(&mut trailing)
.map_err(|e| GitError::InvalidPackFile(format!("Read index error: {e}")))?
!= 0
{
return Err(GitError::InvalidPackFile(
"Pack index has trailing data after checksum".to_string(),
));
}
let pack_file = File::open(pack_path)
.map_err(|e| GitError::InvalidPackFile(format!("Open pack file error: {e}")))?;
let pack_header_file = pack_file
.try_clone()
.map_err(|e| GitError::InvalidPackFile(format!("Open pack file error: {e}")))?;
let mut pack = io::BufReader::new(pack_header_file);
let (header_object_num, _) = Pack::check_header(&mut pack)?;
if header_object_num as usize != object_num {
return Err(GitError::InvalidPackFile(format!(
"Pack index object count {object_num} does not match pack header {header_object_num}"
)));
}
let retention = DecodeRetention::default();
#[cfg(unix)]
{
let scan_threads = thread::available_parallelism()
.map_or(1, usize::from)
.min(objects_by_offset.len())
.min(2);
if scan_threads <= 1 {
Self::scan_object_dependencies_from_index_window(
&pack_file,
&objects_by_offset,
&retention,
)?;
} else {
let chunk_size = objects_by_offset.len().div_ceil(scan_threads);
thread::scope(|scope| {
let mut handles = Vec::with_capacity(scan_threads);
for chunk in objects_by_offset.chunks(chunk_size) {
let pack_file = &pack_file;
let retention = &retention;
handles.push(scope.spawn(move || {
Self::scan_object_dependencies_from_index_window(
pack_file, chunk, retention,
)
}));
}
for handle in handles {
handle.join().map_err(|_| {
GitError::InvalidPackFile("Pack dependency scan panicked".to_string())
})??;
}
Ok::<(), GitError>(())
})?;
}
}
#[cfg(not(unix))]
{
for &(object_offset, _) in &objects_by_offset {
pack.seek(SeekFrom::Start(object_offset as u64))
.map_err(|e| GitError::InvalidPackFile(format!("Read pack file error: {e}")))?;
Self::scan_object_dependency(&mut pack, object_offset, &retention)?;
}
}
let object_hashes = objects_by_offset
.into_iter()
.map(|(_, hash)| hash)
.collect::<Vec<_>>();
Ok(DecodeScan {
retention,
object_hashes: Some(object_hashes),
pack_hash: Some(pack_hash),
pack_hash_check: None,
})
}
#[cfg(unix)]
fn scan_object_dependencies_from_index_window(
pack_file: &File,
objects_by_offset: &[(usize, ObjectHash)],
retention: &DecodeRetention,
) -> Result<(), GitError> {
let mut window = vec![0; PACK_SCAN_WINDOW_SIZE.max(PACK_OBJECT_PREFIX_READ_SIZE)];
let mut window_start = 0usize;
let mut window_len = 0usize;
for &(object_offset, _) in objects_by_offset {
let required_end = object_offset.saturating_add(PACK_OBJECT_PREFIX_READ_SIZE);
let window_end = window_start.saturating_add(window_len);
if window_len == 0 || object_offset < window_start || required_end > window_end {
window_start = object_offset;
window_len = pack_file
.read_at(&mut window, object_offset as u64)
.map_err(|e| GitError::InvalidPackFile(format!("Read pack file error: {e}")))?;
if window_len == 0 {
return Err(GitError::InvalidPackFile(
"Unexpected EOF while scanning pack dependencies".to_string(),
));
}
}
let prefix_start = object_offset - window_start;
let prefix_len = window_len
.saturating_sub(prefix_start)
.min(PACK_OBJECT_PREFIX_READ_SIZE);
if prefix_len == 0 {
return Err(GitError::InvalidPackFile(
"Unexpected EOF while scanning pack dependencies".to_string(),
));
}
let mut object_prefix = Cursor::new(&window[prefix_start..prefix_start + prefix_len]);
Self::scan_object_dependency(&mut object_prefix, object_offset, retention)?;
}
Ok(())
}
fn scan_object_dependency(
pack: &mut impl Read,
init_offset: usize,
retention: &DecodeRetention,
) -> Result<(), GitError> {
let mut offset = init_offset;
let (type_bits, _) = utils::read_type_and_varint_size(pack, &mut offset)
.map_err(|e| GitError::InvalidPackFile(format!("Read error: {e}")))?;
let obj_type = ObjectType::from_pack_type_u8(type_bits)?;
match obj_type {
ObjectType::OffsetDelta | ObjectType::OffsetZstdelta => {
let (delta_offset, _) = utils::read_offset_encoding(pack)
.map_err(|e| GitError::InvalidPackFile(format!("Read error: {e}")))?;
let base_offset =
init_offset
.checked_sub(delta_offset as usize)
.ok_or_else(|| {
GitError::InvalidObjectInfo("Invalid OffsetDelta offset".to_string())
})?;
retention.add_offset_dependency(base_offset);
}
ObjectType::HashDelta => {
let ref_sha = ObjectHash::from_stream(pack)
.map_err(|e| GitError::InvalidPackFile(format!("Read error: {e}")))?;
retention.add_hash_dependency(ref_sha);
}
ObjectType::Commit | ObjectType::Tree | ObjectType::Blob | ObjectType::Tag => {}
other => {
return Err(GitError::InvalidPackFile(format!(
"AI object type `{other}` cannot appear in a pack file"
)));
}
}
Ok(())
}
fn hash_pack_file_payload(pack_path: &Path) -> Result<PackHashCheck, GitError> {
let file = File::open(pack_path)
.map_err(|e| GitError::InvalidPackFile(format!("Open pack file error: {e}")))?;
let len = file
.metadata()
.map_err(|e| GitError::InvalidPackFile(format!("Read pack metadata error: {e}")))?
.len();
let hash_size = get_hash_kind().size();
let hash_size_u64 = hash_size as u64;
if len < hash_size_u64 {
return Err(GitError::InvalidPackFile(
"Pack file is too small to contain a trailer hash".to_string(),
));
}
let mut reader = io::BufReader::with_capacity(FILE_DECODE_BUFFER_SIZE, file);
let mut remaining = len - hash_size_u64;
let mut hasher = HashAlgorithm::new();
let mut scratch = vec![0; FILE_DECODE_BUFFER_SIZE];
while remaining > 0 {
let chunk_len = (remaining as usize).min(scratch.len());
reader
.read_exact(&mut scratch[..chunk_len])
.map_err(|e| GitError::InvalidPackFile(format!("Read pack file error: {e}")))?;
hasher.update(&scratch[..chunk_len]);
remaining -= chunk_len as u64;
}
let mut trailer = vec![0; hash_size];
reader
.read_exact(&mut trailer)
.map_err(|e| GitError::InvalidPackFile(format!("Read pack file error: {e}")))?;
let payload_hash = ObjectHash::from_bytes(&hasher.finalize())
.map_err(|e| GitError::InvalidPackFile(format!("Read pack file error: {e}")))?;
let trailer_hash = ObjectHash::from_bytes(&trailer)
.map_err(|e| GitError::InvalidPackFile(format!("Read pack file error: {e}")))?;
Ok(PackHashCheck {
payload_hash,
trailer_hash,
})
}
fn scan_decode_retention(pack: &mut (impl BufRead + Send)) -> Result<DecodeScan, GitError> {
let (object_num, _) = Pack::check_header(pack)?;
let retention = DecodeRetention::default();
let mut offset: usize = 12;
for _ in 0..object_num {
let init_offset = offset;
let (type_bits, size) = utils::read_type_and_varint_size(pack, &mut offset)
.map_err(|e| GitError::InvalidPackFile(format!("Read error: {e}")))?;
let obj_type = ObjectType::from_pack_type_u8(type_bits)?;
match obj_type {
ObjectType::OffsetDelta | ObjectType::OffsetZstdelta => {
let (delta_offset, bytes) = utils::read_offset_encoding(pack)
.map_err(|e| GitError::InvalidPackFile(format!("Read error: {e}")))?;
offset += bytes;
let base_offset =
init_offset
.checked_sub(delta_offset as usize)
.ok_or_else(|| {
GitError::InvalidObjectInfo(
"Invalid OffsetDelta offset".to_string(),
)
})?;
retention.add_offset_dependency(base_offset);
}
ObjectType::HashDelta => {
let ref_sha = ObjectHash::from_stream(pack)
.map_err(|e| GitError::InvalidPackFile(format!("Read error: {e}")))?;
offset += get_hash_kind().size();
retention.add_hash_dependency(ref_sha);
}
ObjectType::Commit | ObjectType::Tree | ObjectType::Blob | ObjectType::Tag => {}
other => {
return Err(GitError::InvalidPackFile(format!(
"AI object type `{other}` cannot appear in a pack file"
)));
}
}
let raw_size = Pack::skip_compressed_data(pack, size)?;
offset += raw_size;
}
let mut trailer = vec![0; get_hash_kind().size()];
pack.read_exact(&mut trailer)
.map_err(|e| GitError::InvalidPackFile(format!("Read error: {e}")))?;
if !utils::is_eof(pack) {
return Err(GitError::InvalidPackFile(
"The pack file is not at the end".to_string(),
));
}
Ok(DecodeScan {
retention,
object_hashes: None,
pack_hash: None,
pack_hash_check: None,
})
}
fn scan_decode_retention_and_copy(
pack: &mut (impl BufRead + Send),
writer: &mut (impl Write + Send),
) -> Result<DecodeScan, GitError> {
let mut tee = TeeReader {
reader: pack,
writer,
write_error: None,
payload_hash: HashAlgorithm::new(),
hash_tail: Vec::with_capacity(get_hash_kind().size()),
hash_size: get_hash_kind().size(),
};
let mut scan = Pack::scan_decode_retention(&mut tee)?;
tee.check_write_error()
.map_err(|e| GitError::InvalidPackFile(format!("Write temp pack file error: {e}")))?;
scan.pack_hash_check = Some(tee.finish_pack_hash_check()?);
Ok(scan)
}
pub fn decode_pack_object(
pack: &mut (impl BufRead + Send),
offset: &mut usize,
) -> Result<Option<CacheObject>, GitError> {
Self::decode_pack_object_with_crc(pack, offset, true, false, false, None, None)
}
fn decode_pack_object_with_crc(
pack: &mut (impl BufRead + Send),
offset: &mut usize,
track_crc: bool,
skip_unneeded_objects: bool,
emit_skipped_base_callback: bool,
known_hash: Option<ObjectHash>,
retention: Option<&DecodeRetention>,
) -> Result<Option<CacheObject>, GitError> {
let init_offset = *offset;
let mut reader = CrcCountingReader {
inner: pack,
bytes_read: 0,
crc: track_crc.then(crc32fast::Hasher::new),
};
let (type_bits, size) = match utils::read_type_and_varint_size(&mut reader, offset) {
Ok(result) => result,
Err(e) => {
return Err(GitError::InvalidPackFile(format!("Read error: {e}")));
}
};
let t = ObjectType::from_pack_type_u8(type_bits)?;
match t {
ObjectType::Commit | ObjectType::Tree | ObjectType::Blob | ObjectType::Tag => {
if Self::should_skip_no_callback_object(
track_crc,
skip_unneeded_objects,
known_hash,
retention,
init_offset,
) {
let raw_size = Pack::skip_compressed_data(&mut reader, size)?;
*offset += raw_size;
if emit_skipped_base_callback {
let hash = known_hash.ok_or_else(|| {
GitError::InvalidPackFile(
"Missing object hash for skipped callback entry".to_string(),
)
})?;
return Ok(Some(CacheObject {
info: CacheObjectInfo::BaseObject(t, hash),
offset: init_offset,
crc32: 0,
data_decompressed: Vec::new(),
mem_recorder: None,
is_delta_in_pack: false,
known_hash: None,
}));
}
return Ok(None);
}
let (data, raw_size) = Pack::decompress_data(&mut reader, size)?;
*offset += raw_size;
let crc32 = reader.crc32();
let hash = known_hash.unwrap_or_else(|| utils::calculate_object_hash(t, &data));
Ok(Some(CacheObject {
info: CacheObjectInfo::BaseObject(t, hash),
offset: init_offset,
crc32,
data_decompressed: data,
mem_recorder: None,
is_delta_in_pack: false,
known_hash: None,
}))
}
ObjectType::OffsetDelta | ObjectType::OffsetZstdelta => {
let (delta_offset, bytes) =
utils::read_offset_encoding(&mut reader).map_err(|e| {
GitError::InvalidPackFile(format!("Read offset-delta base error: {e}"))
})?;
*offset += bytes;
let delta_offset = usize::try_from(delta_offset).map_err(|_| {
GitError::InvalidObjectInfo("Invalid OffsetDelta offset".to_string())
})?;
let base_offset = init_offset.checked_sub(delta_offset).ok_or_else(|| {
GitError::InvalidObjectInfo("Invalid OffsetDelta offset".to_string())
})?;
if emit_skipped_base_callback
&& Self::should_skip_no_callback_object(
false,
skip_unneeded_objects,
known_hash,
retention,
init_offset,
)
{
let raw_size = Pack::skip_compressed_data(&mut reader, size)?;
*offset += raw_size;
let obj_info = match t {
ObjectType::OffsetDelta => CacheObjectInfo::OffsetDelta(base_offset, 0),
ObjectType::OffsetZstdelta => {
CacheObjectInfo::OffsetZstdelta(base_offset, 0)
}
_ => unreachable!(),
};
return Ok(Some(CacheObject {
info: obj_info,
offset: init_offset,
crc32: 0,
data_decompressed: Vec::new(),
mem_recorder: None,
is_delta_in_pack: true,
known_hash,
}));
}
if !emit_skipped_base_callback
&& Self::should_skip_no_callback_object(
track_crc,
skip_unneeded_objects,
known_hash,
retention,
init_offset,
)
{
let raw_size = Pack::skip_compressed_data(&mut reader, size)?;
*offset += raw_size;
let obj_info = match t {
ObjectType::OffsetDelta => CacheObjectInfo::OffsetDelta(base_offset, 0),
ObjectType::OffsetZstdelta => {
CacheObjectInfo::OffsetZstdelta(base_offset, 0)
}
_ => unreachable!(),
};
return Ok(Some(CacheObject {
info: obj_info,
offset: init_offset,
crc32: 0,
data_decompressed: Vec::new(),
mem_recorder: None,
is_delta_in_pack: true,
known_hash,
}));
}
let (data, raw_size) = Pack::decompress_data(&mut reader, size)?;
*offset += raw_size;
let mut delta_reader = Cursor::new(&data);
let (_, final_size) = utils::read_delta_object_size(&mut delta_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 = reader.crc32();
Ok(Some(CacheObject {
info: obj_info,
offset: init_offset,
crc32,
data_decompressed: data,
mem_recorder: None,
is_delta_in_pack: true,
known_hash,
}))
}
ObjectType::HashDelta => {
let ref_sha = ObjectHash::from_stream(&mut reader).map_err(|e| {
GitError::InvalidPackFile(format!("Read hash-delta base hash error: {e}"))
})?;
*offset += get_hash_kind().size();
if emit_skipped_base_callback
&& Self::should_skip_no_callback_object(
false,
skip_unneeded_objects,
known_hash,
retention,
init_offset,
)
{
let raw_size = Pack::skip_compressed_data(&mut reader, size)?;
*offset += raw_size;
return Ok(Some(CacheObject {
info: CacheObjectInfo::HashDelta(ref_sha, 0),
offset: init_offset,
crc32: 0,
data_decompressed: Vec::new(),
mem_recorder: None,
is_delta_in_pack: true,
known_hash,
}));
}
if !emit_skipped_base_callback
&& Self::should_skip_no_callback_object(
track_crc,
skip_unneeded_objects,
known_hash,
retention,
init_offset,
)
{
let raw_size = Pack::skip_compressed_data(&mut reader, size)?;
*offset += raw_size;
return Ok(Some(CacheObject {
info: CacheObjectInfo::HashDelta(ref_sha, 0),
offset: init_offset,
crc32: 0,
data_decompressed: Vec::new(),
mem_recorder: None,
is_delta_in_pack: true,
known_hash,
}));
}
let (data, raw_size) = Pack::decompress_data(&mut reader, size)?;
*offset += raw_size;
let mut delta_reader = Cursor::new(&data);
let (_, final_size) = utils::read_delta_object_size(&mut delta_reader)?;
let crc32 = reader.crc32();
Ok(Some(CacheObject {
info: CacheObjectInfo::HashDelta(ref_sha, final_size),
offset: init_offset,
crc32,
data_decompressed: data,
mem_recorder: None,
is_delta_in_pack: true,
known_hash,
}))
}
other => Err(GitError::InvalidPackFile(format!(
"AI object type `{other}` cannot appear in a pack file"
))),
}
}
fn should_skip_no_callback_object(
track_crc: bool,
skip_unneeded_objects: bool,
known_hash: Option<ObjectHash>,
retention: Option<&DecodeRetention>,
offset: usize,
) -> bool {
if track_crc || !skip_unneeded_objects {
return false;
}
match (known_hash, retention) {
(Some(hash), Some(retention)) => !retention.should_retain(offset, hash),
_ => false,
}
}
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 callback: DecodeCallback = Arc::new(callback);
if self
.mem_limit
.is_some_and(|limit| limit >= UNBOUNDED_CACHE_THRESHOLD_BYTES)
{
#[cfg(unix)]
if let Some(pack_path) = Self::single_open_pack_path_at_start()
&& Self::reader_matches_pack_prefix(pack, &pack_path)
{
let pack_len = std::fs::metadata(&pack_path)
.map_err(|e| GitError::InvalidPackFile(format!("Read pack file error: {e}")))?
.len();
self.decode_file_inner_with_sync_base_callbacks(
&pack_path,
Some(callback),
pack_id_callback,
if Self::low_memory_callback_entries() {
FileDecodeMode::SkipUnneeded
} else {
FileDecodeMode::RetainAll
},
true,
)?;
Self::consume_reader_exact(pack, pack_len)?;
return Ok(());
}
let mut temp_pack = NamedTempFile::new().map_err(|e| {
GitError::InvalidPackFile(format!("Create temp pack file error: {e}"))
})?;
let scan = {
let mut temp_writer =
io::BufWriter::with_capacity(FILE_DECODE_BUFFER_SIZE, &mut temp_pack);
let scan = Pack::scan_decode_retention_and_copy(pack, &mut temp_writer)?;
temp_writer.flush().map_err(|e| {
GitError::InvalidPackFile(format!("Flush temp pack file error: {e}"))
})?;
scan
};
temp_pack.flush().map_err(|e| {
GitError::InvalidPackFile(format!("Flush temp pack file error: {e}"))
})?;
return self.decode_file_inner_with_scan(
temp_pack.path(),
Some(callback),
pack_id_callback,
if Self::low_memory_callback_entries() {
FileDecodeMode::SkipUnneeded
} else {
FileDecodeMode::RetainAll
},
scan,
true,
);
}
self.decode_inner(
pack,
Some(callback),
pack_id_callback,
DecodeOptions::streaming(),
)
}
#[cfg(unix)]
fn single_open_pack_path_at_start() -> Option<PathBuf> {
fn fd_position_is_start(fd_name: &std::ffi::OsStr) -> bool {
let fdinfo_path = Path::new("/proc/self/fdinfo").join(fd_name);
let Ok(fdinfo) = std::fs::read_to_string(fdinfo_path) else {
return false;
};
fdinfo.lines().any(|line| {
let Some(pos) = line.strip_prefix("pos:") else {
return false;
};
pos.trim() == "0"
})
}
let mut found = None;
let fd_dir = std::fs::read_dir("/proc/self/fd").ok()?;
for entry in fd_dir.flatten() {
if !fd_position_is_start(&entry.file_name()) {
continue;
}
let Ok(path) = std::fs::read_link(entry.path()) else {
continue;
};
if path.extension().and_then(|ext| ext.to_str()) != Some("pack") {
continue;
}
if !path.is_file() || !path.with_extension("idx").is_file() {
continue;
}
if found.replace(path).is_some() {
return None;
}
}
found
}
#[cfg(unix)]
fn reader_matches_pack_prefix(pack: &mut (impl BufRead + Send), pack_path: &Path) -> bool {
let Ok(prefix) = pack.fill_buf() else {
return false;
};
if prefix.is_empty() {
return false;
}
let Ok(file) = File::open(pack_path) else {
return false;
};
let Ok(metadata) = file.metadata() else {
return false;
};
let min_prefix = 4096.min(metadata.len() as usize);
if prefix.len() < min_prefix {
return false;
}
let compare_len = prefix.len().min(FILE_DECODE_BUFFER_SIZE);
let mut file_prefix = vec![0; compare_len];
match file.read_at(&mut file_prefix, 0) {
Ok(n) if n == compare_len => file_prefix == prefix[..compare_len],
_ => false,
}
}
#[cfg(unix)]
fn consume_reader_exact(
pack: &mut (impl BufRead + Send),
mut bytes: u64,
) -> Result<(), GitError> {
while bytes != 0 {
let buf = pack
.fill_buf()
.map_err(|e| GitError::InvalidPackFile(format!("Read pack file error: {e}")))?;
if buf.is_empty() {
return Err(GitError::InvalidPackFile(
"Pack reader ended before matched pack bytes were consumed".to_string(),
));
}
let consumed =
usize::try_from(bytes).map_or(buf.len(), |remaining| remaining.min(buf.len()));
pack.consume(consumed);
bytes -= consumed as u64;
}
Ok(())
}
pub fn decode_without_callback<C>(
&mut self,
pack: &mut (impl BufRead + Send),
pack_id_callback: Option<C>,
) -> Result<(), GitError>
where
C: FnOnce(ObjectHash) + Send + 'static,
{
self.decode_inner(pack, None, pack_id_callback, DecodeOptions::streaming())
}
pub fn decode_file<F, C>(
&mut self,
pack_path: impl AsRef<Path>,
callback: F,
pack_id_callback: Option<C>,
) -> Result<(), GitError>
where
F: Fn(MetaAttached<Entry, EntryMeta>) + Sync + Send + 'static,
C: FnOnce(ObjectHash) + Send + 'static,
{
let callback: DecodeCallback = Arc::new(callback);
self.decode_file_inner(
pack_path.as_ref(),
Some(callback),
pack_id_callback,
FileDecodeMode::RetainAll,
)
}
pub fn decode_file_full_without_callback<C>(
&mut self,
pack_path: impl AsRef<Path>,
pack_id_callback: Option<C>,
) -> Result<(), GitError>
where
C: FnOnce(ObjectHash) + Send + 'static,
{
self.decode_file_inner(
pack_path.as_ref(),
None,
pack_id_callback,
FileDecodeMode::RetainAll,
)
}
pub fn decode_file_without_callback<C>(
&mut self,
pack_path: impl AsRef<Path>,
pack_id_callback: Option<C>,
) -> Result<(), GitError>
where
C: FnOnce(ObjectHash) + Send + 'static,
{
self.decode_file_inner(
pack_path.as_ref(),
None,
pack_id_callback,
FileDecodeMode::SkipUnneeded,
)
}
fn decode_file_inner<C>(
&mut self,
pack_path: &Path,
callback: Option<DecodeCallback>,
pack_id_callback: Option<C>,
mode: FileDecodeMode,
) -> Result<(), GitError>
where
C: FnOnce(ObjectHash) + Send + 'static,
{
self.decode_file_inner_with_sync_base_callbacks(
pack_path,
callback,
pack_id_callback,
mode,
true,
)
}
fn decode_file_inner_with_sync_base_callbacks<C>(
&mut self,
pack_path: &Path,
callback: Option<DecodeCallback>,
pack_id_callback: Option<C>,
mode: FileDecodeMode,
sync_base_callbacks: bool,
) -> Result<(), GitError>
where
C: FnOnce(ObjectHash) + Send + 'static,
{
let scan = match Pack::scan_decode_retention_from_index(pack_path) {
Ok(scan) => scan,
Err(_) => {
let scan_file = File::open(pack_path)
.map_err(|e| GitError::InvalidPackFile(format!("Open pack file error: {e}")))?;
let mut scan_reader = io::BufReader::new(scan_file);
Pack::scan_decode_retention(&mut scan_reader)?
}
};
self.decode_file_inner_with_scan(
pack_path,
callback,
pack_id_callback,
mode,
scan,
sync_base_callbacks,
)
}
fn decode_file_inner_with_scan<C>(
&mut self,
pack_path: &Path,
callback: Option<DecodeCallback>,
pack_id_callback: Option<C>,
mode: FileDecodeMode,
scan: DecodeScan,
sync_base_callbacks: bool,
) -> Result<(), GitError>
where
C: FnOnce(ObjectHash) + Send + 'static,
{
let DecodeScan {
retention,
object_hashes: known_hashes,
pack_hash,
pack_hash_check,
} = scan;
let expected_pack_hash =
pack_hash.or_else(|| pack_hash_check.as_ref().map(|check| check.payload_hash));
let retention = Arc::new(retention);
let retention_mode = match mode {
FileDecodeMode::RetainAll => DecodeRetentionMode::retain_all(retention),
FileDecodeMode::SkipUnneeded => DecodeRetentionMode::skip_unneeded(retention),
};
let skip_payload_hash_check = callback.is_some() && Self::low_memory_callback_entries();
let hash_check =
if !skip_payload_hash_check && pack_hash.is_some() && pack_hash_check.is_none() {
let pack_path = pack_path.to_path_buf();
let kind: HashKind = get_hash_kind();
Some(thread::spawn(move || {
set_hash_kind(kind);
Pack::hash_pack_file_payload(&pack_path)
}))
} else {
None
};
let file = File::open(pack_path)
.map_err(|e| GitError::InvalidPackFile(format!("Open pack file error: {e}")))?;
let mut reader = io::BufReader::with_capacity(FILE_DECODE_BUFFER_SIZE, file);
let decode_result = self.decode_inner(
&mut reader,
callback,
pack_id_callback,
DecodeOptions {
retention_mode,
known_hashes,
expected_pack_hash,
verify_pack_stream_hash: !skip_payload_hash_check
&& hash_check.is_none()
&& pack_hash_check.is_none(),
sync_base_callbacks,
},
);
let hash_result = hash_check.map(|handle| {
handle
.join()
.map_err(|_| GitError::InvalidPackFile("Pack hash check panicked".to_string()))?
});
decode_result?;
if let Some(hash_check) = pack_hash_check {
Self::verify_pack_hash_check(&hash_check, self.signature)?;
} else if let Some(hash_check) = hash_result {
let hash_check = hash_check?;
Self::verify_pack_hash_check(&hash_check, self.signature)?;
}
Ok(())
}
fn verify_pack_hash_check(
hash_check: &PackHashCheck,
signature: ObjectHash,
) -> Result<(), GitError> {
if hash_check.trailer_hash != signature {
return Err(GitError::InvalidPackFile(format!(
"The pack file trailer hash {} does not match decoded trailer hash {}",
hash_check.trailer_hash, signature
)));
}
if hash_check.payload_hash != signature {
return Err(GitError::InvalidPackFile(format!(
"The pack file hash {} does not match the trailer hash {}",
hash_check.payload_hash, signature
)));
}
Ok(())
}
fn decode_inner<C>(
&mut self,
pack: &mut (impl BufRead + Send),
callback: Option<DecodeCallback>,
pack_id_callback: Option<C>,
options: DecodeOptions,
) -> Result<(), GitError>
where
C: FnOnce(ObjectHash) + Send + 'static,
{
let DecodeOptions {
retention_mode,
known_hashes,
expected_pack_hash,
verify_pack_stream_hash,
sync_base_callbacks,
} = options;
let time = Instant::now();
let mut last_update_time = time.elapsed().as_millis();
let log_enabled = tracing::enabled!(tracing::Level::INFO);
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 track_crc = callback.is_some() && !Self::low_memory_callback_entries();
let known_hashes = known_hashes.as_deref();
let shared_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,
retention: retention_mode.retention,
skip_unneeded_objects: retention_mode.skip_unneeded_objects,
});
let mut reader = if verify_pack_stream_hash {
Wrapper::new(pack)
} else {
Wrapper::new_without_hash(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;
let mem_limit = if self.caches.is_unbounded() {
None
} else {
self.mem_limit
};
while i < self.number {
if log_enabled && 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;
}
}
if let Some(mem_limit) = mem_limit {
while self.pool.queued_count() > MAX_QUEUED_DECODE_TASKS
|| self.memory_used() > mem_limit
{
thread::yield_now();
}
} else {
while self.pool.queued_count() > MAX_QUEUED_DECODE_TASKS {
thread::yield_now();
}
}
let known_hash = known_hashes.and_then(|hashes| hashes.get(i).copied());
let r: Result<Option<CacheObject>, GitError> = Pack::decode_pack_object_with_crc(
&mut reader,
&mut offset,
track_crc,
shared_params.skip_unneeded_objects,
shared_params.callback.is_some() && Self::low_memory_callback_entries(),
known_hash,
shared_params.retention.as_deref(),
);
match r {
Ok(Some(obj)) => {
let Some(mut obj) = Self::try_process_skipped_low_memory_callback_object(
&shared_params,
obj,
Self::low_memory_callback_entries(),
) else {
i += 1;
continue;
};
if Self::should_skip_no_callback_delta(&shared_params, &obj) {
Self::process_delta_dependency(shared_params.clone(), obj);
i += 1;
continue;
}
if matches!(obj.info, CacheObjectInfo::BaseObject(_, _))
&& Self::should_drop_no_callback_base(&shared_params, &obj)
{
i += 1;
continue;
}
obj.set_mem_recorder(self.cache_objs_mem.clone());
obj.record_mem_size();
if matches!(obj.info, CacheObjectInfo::BaseObject(_, _))
&& (shared_params.callback.is_none() || sync_base_callbacks)
{
Self::cache_obj_and_process_waitlist(&shared_params, obj);
i += 1;
continue;
}
let params = shared_params.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(¶ms, obj);
}
CacheObjectInfo::OffsetDelta(_, _)
| CacheObjectInfo::OffsetZstdelta(_, _)
| CacheObjectInfo::HashDelta(_, _) => {
Self::process_delta_dependency(params, obj);
}
}
});
}
Ok(None) => {}
Err(e) => {
self.abort_decode();
return Err(e);
}
}
i += 1;
}
log_info(i, self);
let render_hash = verify_pack_stream_hash.then(|| reader.final_hash());
self.signature = match ObjectHash::from_stream(&mut reader) {
Ok(signature) => signature,
Err(e) => {
self.abort_decode();
return Err(GitError::InvalidPackFile(format!(
"Error reading pack trailer hash: {e}"
)));
}
};
if let Some(expected_pack_hash) = expected_pack_hash
&& expected_pack_hash != self.signature
{
self.abort_decode();
return Err(GitError::InvalidPackFile(format!(
"The pack index hash {} does not match the trailer hash {}",
expected_pack_hash, self.signature
)));
}
if let Some(render_hash) = render_hash
&& render_hash != self.signature
{
self.abort_decode();
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 {
self.abort_decode();
return Err(GitError::InvalidPackFile(
"The pack file is not at the end".to_string(),
));
}
self.pool.join();
if let Some(pack_callback) = pack_id_callback {
pack_callback(self.signature);
}
assert_eq!(self.waitlist.map_offset.len(), 0);
assert_eq!(self.waitlist.map_ref.len(), 0);
assert!(self.number >= self.caches.total_inserted());
tracing::info!(
"The pack file has been decoded successfully, takes: [ {:?} ]",
time.elapsed()
);
self.caches.clear(); assert_eq!(self.cache_objs_mem_used(), 0);
Ok(())
}
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
})
}
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;
}
}
});
tokio::task::spawn_blocking(move || {
set_hash_kind(kind);
self.decode(
&mut reader,
move |entry: MetaAttached<Entry, EntryMeta>| {
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()
}
fn memory_used(&self) -> usize {
self.cache_objs_mem_used() + self.caches.memory_used_index()
}
fn cache_objs_mem_used(&self) -> usize {
self.cache_objs_mem.load(Ordering::Acquire)
}
fn release_offset_dependency(
shared_params: &SharedParams,
base_offset: usize,
base_obj: &CacheObject,
) {
if let Some(retention) = &shared_params.retention {
retention.consume_offset_dependency(base_offset);
Self::maybe_remove_released_base(shared_params, base_obj);
}
}
fn release_hash_dependency(
shared_params: &SharedParams,
base_hash: ObjectHash,
base_obj: &CacheObject,
) {
if let Some(retention) = &shared_params.retention {
retention.consume_hash_dependency(base_hash);
Self::maybe_remove_released_base(shared_params, base_obj);
}
}
fn maybe_remove_released_base(shared_params: &SharedParams, base_obj: &CacheObject) {
if let Some(retention) = &shared_params.retention
&& let Some(hash) = base_obj.base_object_hash()
&& !retention.should_retain(base_obj.offset, hash)
{
shared_params.caches.remove_unbounded(base_obj.offset, hash);
}
}
fn process_waitlist_objects(
shared_params: &Arc<SharedParams>,
wait_objs: Vec<CacheObject>,
base_obj: Arc<CacheObject>,
) {
for obj in wait_objs {
Self::process_delta(Arc::clone(shared_params), obj, base_obj.clone());
}
}
fn try_process_skipped_low_memory_callback_object(
shared_params: &Arc<SharedParams>,
obj: CacheObject,
low_memory_callback_entries: bool,
) -> Option<CacheObject> {
if !low_memory_callback_entries || !obj.data_decompressed.is_empty() {
return Some(obj);
}
let (Some(callback), Some(retention)) = (
shared_params.callback.as_ref(),
shared_params.retention.as_ref(),
) else {
return Some(obj);
};
match &obj.info {
CacheObjectInfo::BaseObject(_, hash) => {
if retention.should_retain(obj.offset, *hash)
|| shared_params.waitlist.has_waiters(obj.offset, *hash)
{
return Some(obj);
}
callback(Self::callback_entry_owned(obj));
None
}
CacheObjectInfo::OffsetDelta(_, _)
| CacheObjectInfo::OffsetZstdelta(_, _)
| CacheObjectInfo::HashDelta(_, _) => {
let Some(hash) = obj.known_hash else {
return Some(obj);
};
if retention.should_retain(obj.offset, hash)
|| shared_params.waitlist.has_waiters(obj.offset, hash)
{
return Some(obj);
}
Self::process_delta_dependency(shared_params.clone(), obj);
None
}
}
}
fn should_skip_no_callback_delta(
shared_params: &SharedParams,
delta_obj: &CacheObject,
) -> bool {
if shared_params.callback.is_some() {
return false;
}
if !shared_params.skip_unneeded_objects {
return false;
}
let Some(retention) = &shared_params.retention else {
return false;
};
let Some(hash) = delta_obj.known_hash else {
return false;
};
!retention.should_retain(delta_obj.offset, hash)
&& !shared_params
.waitlist
.map_offset
.contains_key(&delta_obj.offset)
&& !shared_params.waitlist.map_ref.contains_key(&hash)
}
fn should_drop_no_callback_base(shared_params: &SharedParams, base_obj: &CacheObject) -> bool {
if shared_params.callback.is_some() {
return false;
}
let Some(retention) = &shared_params.retention else {
return false;
};
let Some(hash) = base_obj.base_object_hash() else {
return false;
};
!retention.should_retain(base_obj.offset, hash)
&& !shared_params.waitlist.has_waiters(base_obj.offset, hash)
}
fn process_delta_dependency(shared_params: Arc<SharedParams>, obj: CacheObject) {
match obj.info {
CacheObjectInfo::OffsetDelta(base_offset, _)
| CacheObjectInfo::OffsetZstdelta(base_offset, _) => {
if let Some(base_obj) = shared_params.caches.get_by_offset(base_offset) {
Self::release_offset_dependency(&shared_params, base_offset, &base_obj);
Self::process_delta(shared_params, obj, base_obj);
} else {
shared_params.waitlist.insert_offset(base_offset, obj);
if let Some(retention) = &shared_params.retention {
retention.consume_offset_dependency(base_offset);
}
if let Some(base_obj) = shared_params.caches.get_by_offset(base_offset) {
Self::maybe_remove_released_base(&shared_params, &base_obj);
Self::process_waitlist(&shared_params, base_obj);
}
}
}
CacheObjectInfo::HashDelta(base_ref, _) => {
if let Some(base_obj) = shared_params.caches.get_by_hash(base_ref) {
Self::release_hash_dependency(&shared_params, base_ref, &base_obj);
Self::process_delta(shared_params, obj, base_obj);
} else {
shared_params.waitlist.insert_ref(base_ref, obj);
if let Some(retention) = &shared_params.retention {
retention.consume_hash_dependency(base_ref);
}
if let Some(base_obj) = shared_params.caches.get_by_hash(base_ref) {
Self::maybe_remove_released_base(&shared_params, &base_obj);
Self::process_waitlist(&shared_params, base_obj);
}
}
}
CacheObjectInfo::BaseObject(_, _) => unreachable!(),
}
}
fn process_delta(
shared_params: Arc<SharedParams>,
delta_obj: CacheObject,
base_obj: Arc<CacheObject>,
) {
if Self::should_skip_no_callback_delta(&shared_params, &delta_obj) {
return;
}
if Self::try_callback_unneeded_low_memory_delta(&shared_params, &delta_obj, &base_obj) {
return;
}
shared_params.pool.clone().execute(move || {
let known_hash = delta_obj.known_hash;
let mut new_obj = match delta_obj.info {
CacheObjectInfo::OffsetDelta(_, _) | CacheObjectInfo::HashDelta(_, _) => {
Pack::rebuild_delta_with_hash(delta_obj, base_obj, known_hash)
}
CacheObjectInfo::OffsetZstdelta(_, _) => {
Pack::rebuild_zstdelta_with_hash(delta_obj, base_obj, known_hash)
}
_ => 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); });
}
fn try_callback_unneeded_low_memory_delta(
shared_params: &SharedParams,
delta_obj: &CacheObject,
base_obj: &CacheObject,
) -> bool {
if !Self::low_memory_callback_entries() {
return false;
}
let (Some(callback), Some(retention), Some(hash)) = (
shared_params.callback.as_ref(),
shared_params.retention.as_ref(),
delta_obj.known_hash,
) else {
return false;
};
if retention.should_retain(delta_obj.offset, hash)
|| shared_params.waitlist.has_waiters(delta_obj.offset, hash)
{
return false;
}
callback(Self::low_memory_delta_callback_entry(
delta_obj,
base_obj.object_type(),
hash,
));
true
}
fn low_memory_delta_callback_entry(
delta_obj: &CacheObject,
obj_type: ObjectType,
hash: ObjectHash,
) -> MetaAttached<Entry, EntryMeta> {
MetaAttached {
inner: Entry {
obj_type,
data: Vec::new(),
hash,
chain_len: 0,
},
meta: EntryMeta {
pack_offset: Some(delta_obj.offset),
crc32: Some(delta_obj.crc32),
is_delta: Some(delta_obj.is_delta_in_pack),
..Default::default()
},
}
}
fn cache_obj_and_process_waitlist(shared_params: &Arc<SharedParams>, new_obj: CacheObject) {
if let Some(retention) = &shared_params.retention {
let hash = new_obj.base_object_hash().unwrap();
let offset = new_obj.offset;
let should_retain = retention.should_retain(offset, hash);
if should_retain {
if let Some(callback) = &shared_params.callback {
callback(Self::callback_entry_ref(&new_obj));
}
let new_obj = shared_params.caches.insert(offset, hash, new_obj);
let wait_objs = shared_params.waitlist.take(offset, hash);
Self::process_waitlist_objects(shared_params, wait_objs, new_obj);
} else {
let wait_objs = shared_params.waitlist.take(offset, hash);
if !wait_objs.is_empty() {
if let Some(callback) = &shared_params.callback {
callback(Self::callback_entry_ref(&new_obj));
}
Self::process_waitlist_objects(shared_params, wait_objs, Arc::new(new_obj));
} else if let Some(callback) = &shared_params.callback {
callback(Self::callback_entry_owned(new_obj));
}
}
return;
}
if let Some(callback) = &shared_params.callback {
callback(Self::callback_entry_ref(&new_obj));
}
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());
Self::process_waitlist_objects(shared_params, wait_objs, base_obj);
}
pub fn rebuild_delta(delta_obj: CacheObject, base_obj: Arc<CacheObject>) -> CacheObject {
Self::rebuild_delta_with_hash(delta_obj, base_obj, None)
}
fn rebuild_delta_with_hash(
delta_obj: CacheObject,
base_obj: Arc<CacheObject>,
known_hash: Option<ObjectHash>,
) -> 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.as_slice());
let (base_size, result_size) = utils::read_delta_object_size(&mut stream).unwrap();
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 {
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 {
if instruction == 0 {
panic!(
"{}",
GitError::DeltaObjectError(String::from("Invalid data instruction"))
);
}
let start = stream.position() as usize;
let end = start + instruction as usize;
let delta_data = *stream.get_ref();
let data = delta_data.get(start..end).unwrap_or_else(|| {
panic!(
"{}",
GitError::DeltaObjectError("Invalid data instruction".to_string())
)
});
result.extend_from_slice(data);
stream.set_position(end as u64);
} else {
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 {
size = COPY_ZERO_SIZE;
}
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 = known_hash
.unwrap_or_else(|| 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,
known_hash: None,
} }
pub fn rebuild_zstdelta(delta_obj: CacheObject, base_obj: Arc<CacheObject>) -> CacheObject {
Self::rebuild_zstdelta_with_hash(delta_obj, base_obj, None)
}
fn rebuild_zstdelta_with_hash(
delta_obj: CacheObject,
base_obj: Arc<CacheObject>,
known_hash: Option<ObjectHash>,
) -> CacheObject {
let result = zstdelta::apply(&base_obj.data_decompressed, &delta_obj.data_decompressed)
.expect("Failed to apply zstdelta");
let hash = known_hash
.unwrap_or_else(|| 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,
known_hash: None,
} }
}
impl Pack {
pub fn stats_pack(path: PathBuf) -> Result<PackStats, crate::errors::GitError> {
PackStats::analyze(path)
}
}
#[cfg(test)]
mod tests {
use std::{
fs,
io::{BufReader, Cursor, prelude::*},
path::{Path, PathBuf},
sync::{
Arc, Mutex,
atomic::{AtomicUsize, Ordering},
},
};
use flate2::{Compression, write::ZlibEncoder};
use futures_util::TryStreamExt;
use sha1::{Digest, Sha1};
use tempfile::tempdir;
use threadpool::ThreadPool;
use tokio_util::io::ReaderStream;
use crate::{
hash::{HashKind, ObjectHash, get_hash_kind, set_hash_kind_for_test},
internal::{
object::types::ObjectType,
pack::{
Pack,
cache::{_Cache, Caches},
cache_object::{CacheObject, CacheObjectInfo},
test_pack_download::download_pack_file,
tests::init_logger,
utils,
waitlist::Waitlist,
},
},
};
fn pack_test_tmp() -> (tempfile::TempDir, PathBuf) {
let dir = tempdir().unwrap();
let path = dir.path().join(".cache_temp");
(dir, path)
}
const LARGE_PACK_TEST_MEM_LIMIT: usize = super::UNBOUNDED_CACHE_THRESHOLD_BYTES;
#[cfg_attr(coverage, ignore)]
#[ignore = "requires large remote pack fixture"]
#[tokio::test]
async fn test_pack_check_header() {
let (source, _guard) = download_pack_file("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!"; 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();
let mut cursor: Cursor<Vec<u8>> = Cursor::new(compressed_data);
let expected_size = data.len();
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:?}"),
}
}
#[test]
fn test_pack_decode_truncated_pack_returns_err_without_panic() {
let _guard = set_hash_kind_for_test(HashKind::Sha1);
let (source, _dl_guard) = download_pack_file("small-sha1.pack");
let mut bytes = fs::read(source).unwrap();
bytes.truncate(bytes.len() - 1);
let tmp_dir = tempfile::tempdir().unwrap();
let tmp_path = tmp_dir.path().to_path_buf();
let result = std::panic::catch_unwind(std::panic::AssertUnwindSafe(move || {
let mut buffered = BufReader::new(Cursor::new(bytes));
let mut pack = Pack::new(Some(2), Some(1024 * 1024), Some(tmp_path), true);
pack.decode(&mut buffered, |_| {}, None::<fn(ObjectHash)>)
}));
assert!(result.is_ok(), "truncated pack decode should not panic");
assert!(
matches!(
result.unwrap(),
Err(crate::errors::GitError::InvalidPackFile(_))
| Err(crate::errors::GitError::IOError(_))
),
"truncated pack decode should return a pack error"
);
}
#[test]
fn test_skip_compressed_data_exact_size_and_size_errors() {
let data = b"Hello, world!";
let mut encoder = ZlibEncoder::new(Vec::new(), Compression::default());
encoder.write_all(data).unwrap();
let compressed = encoder.finish().unwrap();
let mut exact = Cursor::new(compressed.clone());
let bytes_read = Pack::skip_compressed_data(&mut exact, data.len()).unwrap();
assert_eq!(bytes_read, compressed.len());
let mut too_large = Cursor::new(compressed.clone());
let err = Pack::skip_compressed_data(&mut too_large, data.len() + 1).unwrap_err();
assert!(err.to_string().contains("smaller than the expected size"));
let mut too_small = Cursor::new(compressed);
let err = Pack::skip_compressed_data(&mut too_small, data.len() - 1).unwrap_err();
assert!(err.to_string().contains("exceeds the expected size"));
}
#[test]
fn test_read_be_u64() {
let mut reader = Cursor::new(0x0102_0304_0506_0708u64.to_be_bytes());
assert_eq!(
Pack::read_be_u64(&mut reader).unwrap(),
0x0102_0304_0506_0708
);
}
#[test]
fn test_decode_pack_object_crc_can_be_skipped() {
let _guard = set_hash_kind_for_test(HashKind::Sha1);
let mut encoder = ZlibEncoder::new(Vec::new(), Compression::default());
encoder.write_all(b"a").unwrap();
let compressed = encoder.finish().unwrap();
let mut object_data = Vec::new();
object_data.push(0x31);
object_data.extend_from_slice(&compressed);
let expected_crc = crc32fast::hash(&object_data);
let mut crc_reader = Cursor::new(object_data.clone());
let mut offset = 0;
let with_crc = Pack::decode_pack_object(&mut crc_reader, &mut offset)
.unwrap()
.unwrap();
assert_eq!(with_crc.crc32, expected_crc);
assert_eq!(with_crc.data_decompressed, b"a");
let mut no_crc_reader = Cursor::new(object_data);
let mut offset = 0;
let supplied_hash = ObjectHash::new(b"known-hash-from-idx");
let without_crc = Pack::decode_pack_object_with_crc(
&mut no_crc_reader,
&mut offset,
false,
false,
false,
Some(supplied_hash),
None,
)
.unwrap()
.unwrap();
assert_eq!(without_crc.crc32, 0);
assert_eq!(without_crc.data_decompressed, b"a");
assert_eq!(without_crc.base_object_hash(), Some(supplied_hash));
}
#[test]
fn test_decode_pack_object_can_emit_skipped_base_callback_entry() {
let _guard = set_hash_kind_for_test(HashKind::Sha1);
let mut encoder = ZlibEncoder::new(Vec::new(), Compression::default());
encoder.write_all(b"a").unwrap();
let compressed = encoder.finish().unwrap();
let mut object_data = Vec::new();
object_data.push(0x31);
object_data.extend_from_slice(&compressed);
let supplied_hash = ObjectHash::new(b"known-hash-from-idx");
let retention = super::DecodeRetention::default();
let mut reader = Cursor::new(object_data);
let mut offset = 0;
let skipped = Pack::decode_pack_object_with_crc(
&mut reader,
&mut offset,
false,
true,
true,
Some(supplied_hash),
Some(&retention),
)
.unwrap()
.unwrap();
assert_eq!(skipped.crc32, 0);
assert!(skipped.data_decompressed.is_empty());
assert_eq!(skipped.base_object_hash(), Some(supplied_hash));
assert_eq!(offset, reader.get_ref().len());
}
#[test]
fn test_decode_pack_object_can_skip_unneeded_delta_payload() {
let _guard = set_hash_kind_for_test(HashKind::Sha1);
let mut encoder = ZlibEncoder::new(Vec::new(), Compression::default());
encoder.write_all(b"abc").unwrap();
let compressed = encoder.finish().unwrap();
let mut object_data = Vec::new();
object_data.push(0x63);
object_data.push(5);
object_data.extend_from_slice(&compressed);
let supplied_hash = ObjectHash::new(b"known-leaf-delta-hash");
let retention = super::DecodeRetention::default();
let mut reader = Cursor::new(object_data);
let init_offset = 20;
let mut offset = init_offset;
let skipped = Pack::decode_pack_object_with_crc(
&mut reader,
&mut offset,
false,
true,
true,
Some(supplied_hash),
Some(&retention),
)
.unwrap()
.unwrap();
assert_eq!(skipped.crc32, 0);
assert!(skipped.data_decompressed.is_empty());
assert_eq!(skipped.known_hash, Some(supplied_hash));
assert_eq!(
skipped.info,
CacheObjectInfo::OffsetDelta(init_offset - 5, 0)
);
assert_eq!(offset, init_offset + reader.get_ref().len());
}
#[test]
fn test_low_memory_delta_callback_entry_uses_known_hash_without_payload() {
let hash = ObjectHash::new(b"known-leaf-delta-hash");
let delta_obj = CacheObject {
info: CacheObjectInfo::OffsetDelta(12, 5),
offset: 40,
crc32: 1234,
data_decompressed: b"delta instructions".to_vec(),
mem_recorder: None,
is_delta_in_pack: true,
known_hash: Some(hash),
};
let entry = Pack::low_memory_delta_callback_entry(&delta_obj, ObjectType::Blob, hash);
assert_eq!(entry.inner.obj_type, ObjectType::Blob);
assert!(entry.inner.data.is_empty());
assert_eq!(entry.inner.hash, hash);
assert_eq!(entry.meta.pack_offset, Some(40));
assert_eq!(entry.meta.crc32, Some(1234));
assert_eq!(entry.meta.is_delta, Some(true));
}
#[test]
fn test_skipped_low_memory_base_callbacks_without_cache_insert() {
let hash = ObjectHash::new(b"known-leaf-base-hash");
let seen = Arc::new(Mutex::new(Vec::new()));
let callback_seen = Arc::clone(&seen);
let callback: super::DecodeCallback = Arc::new(move |entry| {
callback_seen.lock().unwrap().push((
entry.inner.hash,
entry.inner.data.len(),
entry.meta.pack_offset,
));
});
let (_dir, cache_path) = pack_test_tmp();
let shared_params = Arc::new(super::SharedParams {
pool: Arc::new(ThreadPool::new(1)),
waitlist: Arc::new(Waitlist::new()),
caches: Arc::new(Caches::new(None, cache_path, 1)),
cache_objs_mem_size: Arc::new(AtomicUsize::new(0)),
callback: Some(callback),
retention: Some(Arc::new(super::DecodeRetention::default())),
skip_unneeded_objects: true,
});
let obj = CacheObject {
info: CacheObjectInfo::BaseObject(ObjectType::Blob, hash),
offset: 64,
crc32: 0,
data_decompressed: Vec::new(),
mem_recorder: None,
is_delta_in_pack: false,
known_hash: None,
};
let remaining =
Pack::try_process_skipped_low_memory_callback_object(&shared_params, obj, true);
assert!(remaining.is_none());
assert_eq!(shared_params.caches.total_inserted(), 0);
assert_eq!(seen.lock().unwrap().as_slice(), &[(hash, 0, Some(64))]);
}
#[cfg(unix)]
#[test]
fn test_large_mem_decode_ignores_unrelated_open_pack_fd() {
let _guard = set_hash_kind_for_test(HashKind::Sha1);
let dir = tempdir().unwrap();
let (open_pack_data, open_hash) = single_blob_pack(b"open pack data");
let open_pack_path = dir.path().join("open.pack");
fs::write(&open_pack_path, open_pack_data).unwrap();
write_test_idx(&open_pack_path, vec![(open_hash, 12)]);
let _open_pack = fs::File::open(&open_pack_path).unwrap();
let (reader_pack_data, _) = single_blob_pack(b"reader data");
let mut reader = Cursor::new(reader_pack_data);
let decoded = Arc::new(Mutex::new(Vec::new()));
let decoded_for_callback = Arc::clone(&decoded);
let mut pack = Pack::new(
Some(1),
Some(super::UNBOUNDED_CACHE_THRESHOLD_BYTES),
Some(dir.path().join("tmp")),
true,
);
pack.decode(
&mut reader,
move |entry| decoded_for_callback.lock().unwrap().push(entry.inner.data),
None::<fn(ObjectHash)>,
)
.unwrap();
assert_eq!(*decoded.lock().unwrap(), vec![b"reader data".to_vec()]);
}
#[cfg(unix)]
#[test]
fn test_consume_reader_exact_leaves_following_bytes() {
let mut reader = Cursor::new(b"pack-datafollowing-data".to_vec());
Pack::consume_reader_exact(&mut reader, b"pack-data".len() as u64).unwrap();
let mut rest = Vec::new();
reader.read_to_end(&mut rest).unwrap();
assert_eq!(rest, b"following-data");
}
#[test]
fn test_pack_decode_without_callback_empty_pack() {
let _guard = set_hash_kind_for_test(HashKind::Sha1);
let mut pack_data = Vec::new();
pack_data.extend_from_slice(b"PACK");
pack_data.extend_from_slice(&2u32.to_be_bytes());
pack_data.extend_from_slice(&0u32.to_be_bytes());
let trailer = Sha1::digest(&pack_data);
pack_data.extend_from_slice(&trailer);
let mut reader = Cursor::new(pack_data);
let mut pack = Pack::new(Some(1), None, None, true);
pack.decode_without_callback(&mut reader, None::<fn(ObjectHash)>)
.unwrap();
assert_eq!(pack.number, 0);
}
#[test]
fn test_pack_decode_without_callback_single_blob_pack() {
let _guard = set_hash_kind_for_test(HashKind::Sha1);
let mut encoder = ZlibEncoder::new(Vec::new(), Compression::default());
encoder.write_all(b"a").unwrap();
let compressed = encoder.finish().unwrap();
let mut pack_data = Vec::new();
pack_data.extend_from_slice(b"PACK");
pack_data.extend_from_slice(&2u32.to_be_bytes());
pack_data.extend_from_slice(&1u32.to_be_bytes());
pack_data.push(0x31);
pack_data.extend_from_slice(&compressed);
let trailer = Sha1::digest(&pack_data);
pack_data.extend_from_slice(&trailer);
let mut reader = Cursor::new(pack_data);
let mut pack = Pack::new(Some(1), None, None, true);
pack.decode_without_callback(&mut reader, None::<fn(ObjectHash)>)
.unwrap();
assert_eq!(pack.number, 1);
assert_eq!(pack.signature.to_string(), hex::encode(trailer));
}
#[test]
fn test_pack_decode_large_mem_limit_uses_temp_retention_path() {
let _guard = set_hash_kind_for_test(HashKind::Sha1);
let mut encoder = ZlibEncoder::new(Vec::new(), Compression::default());
encoder.write_all(b"a").unwrap();
let compressed = encoder.finish().unwrap();
let mut pack_data = Vec::new();
pack_data.extend_from_slice(b"PACK");
pack_data.extend_from_slice(&2u32.to_be_bytes());
pack_data.extend_from_slice(&1u32.to_be_bytes());
pack_data.push(0x31);
pack_data.extend_from_slice(&compressed);
let trailer = Sha1::digest(&pack_data);
pack_data.extend_from_slice(&trailer);
let seen = Arc::new(Mutex::new(Vec::new()));
let seen_cb = Arc::clone(&seen);
let mut reader = Cursor::new(pack_data);
let mut pack = Pack::new(
Some(1),
Some(super::UNBOUNDED_CACHE_THRESHOLD_BYTES),
None,
true,
);
pack.decode(
&mut reader,
move |entry| seen_cb.lock().unwrap().push(entry.inner.data),
None::<fn(ObjectHash)>,
)
.unwrap();
assert_eq!(pack.number, 1);
assert_eq!(pack.signature.to_string(), hex::encode(trailer));
assert_eq!(*seen.lock().unwrap(), vec![b"a".to_vec()]);
}
#[test]
fn test_pack_decode_file_without_callback_uses_idx() {
let _guard = set_hash_kind_for_test(HashKind::Sha1);
let mut encoder = ZlibEncoder::new(Vec::new(), Compression::default());
encoder.write_all(b"a").unwrap();
let compressed = encoder.finish().unwrap();
let mut pack_data = Vec::new();
pack_data.extend_from_slice(b"PACK");
pack_data.extend_from_slice(&2u32.to_be_bytes());
pack_data.extend_from_slice(&1u32.to_be_bytes());
pack_data.push(0x31);
pack_data.extend_from_slice(&compressed);
let trailer = Sha1::digest(&pack_data);
pack_data.extend_from_slice(&trailer);
let dir = tempdir().unwrap();
let pack_path = dir.path().join("single.pack");
fs::write(&pack_path, &pack_data).unwrap();
let obj_hash = utils::calculate_object_hash(ObjectType::Blob, b"a");
let mut idx_data = Vec::new();
idx_data.extend_from_slice(&0xff74_4f63u32.to_be_bytes());
idx_data.extend_from_slice(&2u32.to_be_bytes());
let first = obj_hash.as_ref()[0] as usize;
for fanout_idx in 0..256 {
let count = if fanout_idx >= first { 1u32 } else { 0u32 };
idx_data.extend_from_slice(&count.to_be_bytes());
}
idx_data.extend_from_slice(obj_hash.as_ref());
idx_data.extend_from_slice(&0u32.to_be_bytes());
idx_data.extend_from_slice(&12u32.to_be_bytes());
append_test_idx_trailer(&mut idx_data, &trailer);
fs::write(pack_path.with_extension("idx"), idx_data).unwrap();
let mut pack = Pack::new(Some(1), None, Some(dir.path().join("tmp")), true);
pack.decode_file_without_callback(&pack_path, None::<fn(ObjectHash)>)
.unwrap();
assert_eq!(pack.number, 1);
assert_eq!(pack.signature.to_string(), hex::encode(trailer));
}
#[test]
fn test_pack_decode_file_callback_uses_idx_and_crc() {
let _guard = set_hash_kind_for_test(HashKind::Sha1);
let mut encoder = ZlibEncoder::new(Vec::new(), Compression::default());
encoder.write_all(b"a").unwrap();
let compressed = encoder.finish().unwrap();
let mut object_data = Vec::new();
object_data.push(0x31);
object_data.extend_from_slice(&compressed);
let expected_crc = crc32fast::hash(&object_data);
let mut pack_data = Vec::new();
pack_data.extend_from_slice(b"PACK");
pack_data.extend_from_slice(&2u32.to_be_bytes());
pack_data.extend_from_slice(&1u32.to_be_bytes());
pack_data.extend_from_slice(&object_data);
let trailer = Sha1::digest(&pack_data);
pack_data.extend_from_slice(&trailer);
let dir = tempdir().unwrap();
let pack_path = dir.path().join("single-callback.pack");
fs::write(&pack_path, &pack_data).unwrap();
let obj_hash = utils::calculate_object_hash(ObjectType::Blob, b"a");
let mut idx_data = Vec::new();
idx_data.extend_from_slice(&0xff74_4f63u32.to_be_bytes());
idx_data.extend_from_slice(&2u32.to_be_bytes());
let first = obj_hash.as_ref()[0] as usize;
for fanout_idx in 0..256 {
let count = if fanout_idx >= first { 1u32 } else { 0u32 };
idx_data.extend_from_slice(&count.to_be_bytes());
}
idx_data.extend_from_slice(obj_hash.as_ref());
idx_data.extend_from_slice(&expected_crc.to_be_bytes());
idx_data.extend_from_slice(&12u32.to_be_bytes());
append_test_idx_trailer(&mut idx_data, &trailer);
fs::write(pack_path.with_extension("idx"), idx_data).unwrap();
let entries = Arc::new(std::sync::Mutex::new(Vec::new()));
let entries_for_cb = Arc::clone(&entries);
let mut pack = Pack::new(Some(1), None, Some(dir.path().join("tmp")), true);
pack.decode_file(
&pack_path,
move |entry| entries_for_cb.lock().unwrap().push(entry),
None::<fn(ObjectHash)>,
)
.unwrap();
let entries = Arc::try_unwrap(entries).unwrap().into_inner().unwrap();
assert_eq!(entries.len(), 1);
assert_eq!(entries[0].inner.hash, obj_hash);
assert_eq!(entries[0].inner.data, b"a");
assert_eq!(entries[0].meta.pack_offset, Some(12));
assert_eq!(entries[0].meta.crc32, Some(expected_crc));
assert_eq!(pack.number, 1);
assert_eq!(pack.signature.to_string(), hex::encode(trailer));
}
#[test]
fn test_pack_decode_file_callback_ignores_idx_with_bad_checksum() {
let _guard = set_hash_kind_for_test(HashKind::Sha1);
let mut encoder = ZlibEncoder::new(Vec::new(), Compression::default());
encoder.write_all(b"a").unwrap();
let compressed = encoder.finish().unwrap();
let mut pack_data = Vec::new();
pack_data.extend_from_slice(b"PACK");
pack_data.extend_from_slice(&2u32.to_be_bytes());
pack_data.extend_from_slice(&1u32.to_be_bytes());
pack_data.push(0x31);
pack_data.extend_from_slice(&compressed);
let trailer = Sha1::digest(&pack_data);
pack_data.extend_from_slice(&trailer);
let dir = tempdir().unwrap();
let pack_path = dir.path().join("bad-idx-checksum.pack");
fs::write(&pack_path, &pack_data).unwrap();
let obj_hash = utils::calculate_object_hash(ObjectType::Blob, b"a");
let mut idx_data = Vec::new();
idx_data.extend_from_slice(&0xff74_4f63u32.to_be_bytes());
idx_data.extend_from_slice(&2u32.to_be_bytes());
let first = obj_hash.as_ref()[0] as usize;
for fanout_idx in 0..256 {
let count = if fanout_idx >= first { 1u32 } else { 0u32 };
idx_data.extend_from_slice(&count.to_be_bytes());
}
let hash_offset = idx_data.len();
idx_data.extend_from_slice(obj_hash.as_ref());
idx_data.extend_from_slice(&0u32.to_be_bytes());
idx_data.extend_from_slice(&12u32.to_be_bytes());
append_test_idx_trailer(&mut idx_data, &trailer);
idx_data[hash_offset] ^= 0xff;
fs::write(pack_path.with_extension("idx"), idx_data).unwrap();
let entries = Arc::new(std::sync::Mutex::new(Vec::new()));
let entries_for_cb = Arc::clone(&entries);
let mut pack = Pack::new(Some(1), None, Some(dir.path().join("tmp")), true);
pack.decode_file(
&pack_path,
move |entry| entries_for_cb.lock().unwrap().push(entry.inner.hash),
None::<fn(ObjectHash)>,
)
.unwrap();
assert_eq!(entries.lock().unwrap().as_slice(), &[obj_hash]);
}
#[test]
fn test_pack_decode_file_full_without_callback_uses_idx() {
let _guard = set_hash_kind_for_test(HashKind::Sha1);
let mut encoder = ZlibEncoder::new(Vec::new(), Compression::default());
encoder.write_all(b"a").unwrap();
let compressed = encoder.finish().unwrap();
let mut pack_data = Vec::new();
pack_data.extend_from_slice(b"PACK");
pack_data.extend_from_slice(&2u32.to_be_bytes());
pack_data.extend_from_slice(&1u32.to_be_bytes());
pack_data.push(0x31);
pack_data.extend_from_slice(&compressed);
let trailer = Sha1::digest(&pack_data);
pack_data.extend_from_slice(&trailer);
let dir = tempdir().unwrap();
let pack_path = dir.path().join("single-full-no-callback.pack");
fs::write(&pack_path, &pack_data).unwrap();
let obj_hash = utils::calculate_object_hash(ObjectType::Blob, b"a");
let mut idx_data = Vec::new();
idx_data.extend_from_slice(&0xff74_4f63u32.to_be_bytes());
idx_data.extend_from_slice(&2u32.to_be_bytes());
let first = obj_hash.as_ref()[0] as usize;
for fanout_idx in 0..256 {
let count = if fanout_idx >= first { 1u32 } else { 0u32 };
idx_data.extend_from_slice(&count.to_be_bytes());
}
idx_data.extend_from_slice(obj_hash.as_ref());
idx_data.extend_from_slice(&0u32.to_be_bytes());
idx_data.extend_from_slice(&12u32.to_be_bytes());
append_test_idx_trailer(&mut idx_data, &trailer);
fs::write(pack_path.with_extension("idx"), idx_data).unwrap();
let mut pack = Pack::new(Some(1), None, Some(dir.path().join("tmp")), true);
pack.decode_file_full_without_callback(&pack_path, None::<fn(ObjectHash)>)
.unwrap();
assert_eq!(pack.number, 1);
assert_eq!(pack.signature.to_string(), hex::encode(trailer));
}
#[test]
fn test_pack_decode_file_full_without_callback_uses_idx_large_offset() {
let _guard = set_hash_kind_for_test(HashKind::Sha1);
let mut encoder = ZlibEncoder::new(Vec::new(), Compression::default());
encoder.write_all(b"a").unwrap();
let compressed = encoder.finish().unwrap();
let mut pack_data = Vec::new();
pack_data.extend_from_slice(b"PACK");
pack_data.extend_from_slice(&2u32.to_be_bytes());
pack_data.extend_from_slice(&1u32.to_be_bytes());
pack_data.push(0x31);
pack_data.extend_from_slice(&compressed);
let trailer = Sha1::digest(&pack_data);
pack_data.extend_from_slice(&trailer);
let dir = tempdir().unwrap();
let pack_path = dir.path().join("single-full-no-callback-large-offset.pack");
fs::write(&pack_path, &pack_data).unwrap();
let obj_hash = utils::calculate_object_hash(ObjectType::Blob, b"a");
let mut idx_data = Vec::new();
idx_data.extend_from_slice(&0xff74_4f63u32.to_be_bytes());
idx_data.extend_from_slice(&2u32.to_be_bytes());
let first = obj_hash.as_ref()[0] as usize;
for fanout_idx in 0..256 {
let count = if fanout_idx >= first { 1u32 } else { 0u32 };
idx_data.extend_from_slice(&count.to_be_bytes());
}
idx_data.extend_from_slice(obj_hash.as_ref());
idx_data.extend_from_slice(&0u32.to_be_bytes());
idx_data.extend_from_slice(&0x8000_0000u32.to_be_bytes());
idx_data.extend_from_slice(&12u64.to_be_bytes());
append_test_idx_trailer(&mut idx_data, &trailer);
fs::write(pack_path.with_extension("idx"), idx_data).unwrap();
let mut pack = Pack::new(Some(1), None, Some(dir.path().join("tmp")), true);
pack.decode_file_full_without_callback(&pack_path, None::<fn(ObjectHash)>)
.unwrap();
assert_eq!(pack.number, 1);
assert_eq!(pack.signature.to_string(), hex::encode(trailer));
}
#[test]
fn test_pack_decode_file_without_callback_falls_back_without_idx() {
let _guard = set_hash_kind_for_test(HashKind::Sha1);
let mut encoder = ZlibEncoder::new(Vec::new(), Compression::default());
encoder.write_all(b"a").unwrap();
let compressed = encoder.finish().unwrap();
let mut pack_data = Vec::new();
pack_data.extend_from_slice(b"PACK");
pack_data.extend_from_slice(&2u32.to_be_bytes());
pack_data.extend_from_slice(&1u32.to_be_bytes());
pack_data.push(0x31);
pack_data.extend_from_slice(&compressed);
let trailer = Sha1::digest(&pack_data);
pack_data.extend_from_slice(&trailer);
let dir = tempdir().unwrap();
let pack_path = dir.path().join("single-no-idx.pack");
fs::write(&pack_path, &pack_data).unwrap();
let mut pack = Pack::new(Some(1), None, Some(dir.path().join("tmp")), true);
pack.decode_file_without_callback(&pack_path, None::<fn(ObjectHash)>)
.unwrap();
assert_eq!(pack.number, 1);
assert_eq!(pack.signature.to_string(), hex::encode(trailer));
}
#[test]
fn test_pack_decode_file_without_callback_rejects_idx_pack_hash_mismatch() {
let _guard = set_hash_kind_for_test(HashKind::Sha1);
let mut encoder = ZlibEncoder::new(Vec::new(), Compression::default());
encoder.write_all(b"a").unwrap();
let compressed = encoder.finish().unwrap();
let mut pack_data = Vec::new();
pack_data.extend_from_slice(b"PACK");
pack_data.extend_from_slice(&2u32.to_be_bytes());
pack_data.extend_from_slice(&1u32.to_be_bytes());
pack_data.push(0x31);
pack_data.extend_from_slice(&compressed);
let trailer = Sha1::digest(&pack_data);
pack_data.extend_from_slice(&trailer);
let dir = tempdir().unwrap();
let pack_path = dir.path().join("bad-pack-hash.pack");
fs::write(&pack_path, &pack_data).unwrap();
let obj_hash = utils::calculate_object_hash(ObjectType::Blob, b"a");
let mut idx_data = Vec::new();
idx_data.extend_from_slice(&0xff74_4f63u32.to_be_bytes());
idx_data.extend_from_slice(&2u32.to_be_bytes());
let first = obj_hash.as_ref()[0] as usize;
for fanout_idx in 0..256 {
let count = if fanout_idx >= first { 1u32 } else { 0u32 };
idx_data.extend_from_slice(&count.to_be_bytes());
}
idx_data.extend_from_slice(obj_hash.as_ref());
idx_data.extend_from_slice(&0u32.to_be_bytes());
idx_data.extend_from_slice(&12u32.to_be_bytes());
append_test_idx_trailer(&mut idx_data, &[0xff; 20]);
fs::write(pack_path.with_extension("idx"), idx_data).unwrap();
let mut pack = Pack::new(Some(1), None, Some(dir.path().join("tmp")), true);
let err = pack
.decode_file_without_callback(&pack_path, None::<fn(ObjectHash)>)
.unwrap_err();
assert!(err.to_string().contains("does not match the trailer hash"));
}
#[test]
fn test_pack_decode_file_full_without_callback_rejects_stale_idx_after_pack_change() {
let _guard = set_hash_kind_for_test(HashKind::Sha1);
let mut encoder = ZlibEncoder::new(Vec::new(), Compression::default());
encoder.write_all(b"a").unwrap();
let original_compressed = encoder.finish().unwrap();
let mut original_pack_data = Vec::new();
original_pack_data.extend_from_slice(b"PACK");
original_pack_data.extend_from_slice(&2u32.to_be_bytes());
original_pack_data.extend_from_slice(&1u32.to_be_bytes());
original_pack_data.push(0x31);
original_pack_data.extend_from_slice(&original_compressed);
let original_trailer = Sha1::digest(&original_pack_data);
let mut encoder = ZlibEncoder::new(Vec::new(), Compression::default());
encoder.write_all(b"b").unwrap();
let changed_compressed = encoder.finish().unwrap();
assert_eq!(original_compressed.len(), changed_compressed.len());
let mut changed_pack_data = Vec::new();
changed_pack_data.extend_from_slice(b"PACK");
changed_pack_data.extend_from_slice(&2u32.to_be_bytes());
changed_pack_data.extend_from_slice(&1u32.to_be_bytes());
changed_pack_data.push(0x31);
changed_pack_data.extend_from_slice(&changed_compressed);
changed_pack_data.extend_from_slice(&original_trailer);
let dir = tempdir().unwrap();
let pack_path = dir.path().join("stale-idx.pack");
fs::write(&pack_path, &changed_pack_data).unwrap();
let obj_hash = utils::calculate_object_hash(ObjectType::Blob, b"a");
let mut idx_data = Vec::new();
idx_data.extend_from_slice(&0xff74_4f63u32.to_be_bytes());
idx_data.extend_from_slice(&2u32.to_be_bytes());
let first = obj_hash.as_ref()[0] as usize;
for fanout_idx in 0..256 {
let count = if fanout_idx >= first { 1u32 } else { 0u32 };
idx_data.extend_from_slice(&count.to_be_bytes());
}
idx_data.extend_from_slice(obj_hash.as_ref());
idx_data.extend_from_slice(&0u32.to_be_bytes());
idx_data.extend_from_slice(&12u32.to_be_bytes());
append_test_idx_trailer(&mut idx_data, &original_trailer);
fs::write(pack_path.with_extension("idx"), idx_data).unwrap();
let mut pack = Pack::new(Some(1), None, Some(dir.path().join("tmp")), true);
let err = pack
.decode_file_full_without_callback(&pack_path, None::<fn(ObjectHash)>)
.unwrap_err();
assert!(err.to_string().contains("does not match the trailer hash"));
}
fn append_compressed(buf: &mut Vec<u8>, data: &[u8]) {
let mut encoder = ZlibEncoder::new(Vec::new(), Compression::default());
encoder.write_all(data).unwrap();
buf.extend_from_slice(&encoder.finish().unwrap());
}
fn single_blob_pack(data: &[u8]) -> (Vec<u8>, ObjectHash) {
let obj_hash = utils::calculate_object_hash(ObjectType::Blob, data);
let mut pack_data = Vec::new();
pack_data.extend_from_slice(b"PACK");
pack_data.extend_from_slice(&2u32.to_be_bytes());
pack_data.extend_from_slice(&1u32.to_be_bytes());
pack_data.push(0x30 | data.len() as u8);
append_compressed(&mut pack_data, data);
let trailer = Sha1::digest(&pack_data);
pack_data.extend_from_slice(&trailer);
(pack_data, obj_hash)
}
fn append_test_idx_trailer(idx_data: &mut Vec<u8>, pack_hash: &[u8]) {
idx_data.extend_from_slice(pack_hash);
let mut idx_hash = crate::utils::HashAlgorithm::new();
idx_hash.update(idx_data);
idx_data.extend_from_slice(&idx_hash.finalize());
}
fn write_test_idx(pack_path: &Path, mut objects: Vec<(ObjectHash, u32)>) {
objects.sort_by(|a, b| a.0.as_ref().cmp(b.0.as_ref()));
let mut idx_data = Vec::new();
idx_data.extend_from_slice(&0xff74_4f63u32.to_be_bytes());
idx_data.extend_from_slice(&2u32.to_be_bytes());
for fanout_idx in 0..256 {
let count = objects
.iter()
.filter(|(hash, _)| hash.as_ref()[0] as usize <= fanout_idx)
.count() as u32;
idx_data.extend_from_slice(&count.to_be_bytes());
}
for (hash, _) in &objects {
idx_data.extend_from_slice(hash.as_ref());
}
for _ in &objects {
idx_data.extend_from_slice(&0u32.to_be_bytes());
}
for (_, offset) in &objects {
idx_data.extend_from_slice(&offset.to_be_bytes());
}
let pack_data = fs::read(pack_path).unwrap();
let hash_size = get_hash_kind().size();
append_test_idx_trailer(&mut idx_data, &pack_data[pack_data.len() - hash_size..]);
fs::write(pack_path.with_extension("idx"), idx_data).unwrap();
}
#[test]
fn test_pack_decode_file_without_callback_releases_delta_bases() {
let _guard = set_hash_kind_for_test(HashKind::Sha1);
let base_hash = utils::calculate_object_hash(ObjectType::Blob, b"hello");
let ofs_hash = utils::calculate_object_hash(ObjectType::Blob, b"hi there");
let ref_hash = utils::calculate_object_hash(ObjectType::Blob, b"HELLO");
let mut pack_data = Vec::new();
pack_data.extend_from_slice(b"PACK");
pack_data.extend_from_slice(&2u32.to_be_bytes());
pack_data.extend_from_slice(&3u32.to_be_bytes());
let base_offset = pack_data.len() as u32;
pack_data.push(0x35);
append_compressed(&mut pack_data, b"hello");
let ofs_offset = pack_data.len() as u32;
pack_data.push(0x6b);
pack_data.push((ofs_offset - base_offset) as u8);
append_compressed(
&mut pack_data,
[b"\x05\x08\x08".as_ref(), b"hi there"].concat().as_slice(),
);
let ref_offset = pack_data.len() as u32;
pack_data.push(0x78);
pack_data.extend_from_slice(base_hash.as_ref());
append_compressed(
&mut pack_data,
[b"\x05\x05\x05".as_ref(), b"HELLO"].concat().as_slice(),
);
let trailer = Sha1::digest(&pack_data);
pack_data.extend_from_slice(&trailer);
let dir = tempdir().unwrap();
let pack_path = dir.path().join("delta.pack");
fs::write(&pack_path, &pack_data).unwrap();
write_test_idx(
&pack_path,
vec![
(base_hash, base_offset),
(ofs_hash, ofs_offset),
(ref_hash, ref_offset),
],
);
let mut pack = Pack::new(Some(1), None, Some(dir.path().join("tmp")), true);
pack.decode_file_without_callback(&pack_path, None::<fn(ObjectHash)>)
.unwrap();
assert_eq!(pack.number, 3);
assert_eq!(pack.signature.to_string(), hex::encode(trailer));
}
#[test]
fn test_rebuild_delta_literal_instruction() {
let _guard = set_hash_kind_for_test(HashKind::Sha1);
let base = Arc::new(CacheObject::new_for_undeltified(
ObjectType::Blob,
b"hello".to_vec(),
12,
0,
));
let delta = CacheObject {
info: CacheObjectInfo::OffsetDelta(12, 8),
offset: 20,
crc32: 0,
data_decompressed: [b"\x05\x08\x08".as_ref(), b"hi there"].concat(),
mem_recorder: None,
is_delta_in_pack: true,
known_hash: None,
};
let rebuilt = Pack::rebuild_delta(delta, base);
assert_eq!(rebuilt.object_type(), ObjectType::Blob);
assert_eq!(rebuilt.data_decompressed, b"hi there");
}
#[test]
#[cfg(target_pointer_width = "32")]
fn test_pack_new_mem_limit_no_overflow_32bit() {
let mem_limit = 1_200_000_000usize;
let (_tmp_dir, tmp) = pack_test_tmp();
let result = std::panic::catch_unwind(|| {
let _p = Pack::new(Some(1), Some(mem_limit), Some(tmp), true);
});
assert!(result.is_ok(), "Pack::new should not panic on 32-bit");
}
fn run_decode_no_delta(filename: &str, kind: HashKind) {
let _guard = set_hash_kind_for_test(kind);
let (source, _dl_guard) = download_pack_file(filename);
let (_tmp_dir, tmp) = pack_test_tmp();
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("small-sha1.pack", HashKind::Sha1);
run_decode_no_delta("small-sha256.pack", HashKind::Sha256);
}
fn run_decode_with_ref_delta(filename: &str, kind: HashKind) {
let _guard = set_hash_kind_for_test(kind);
init_logger();
let (source, _dl_guard) = download_pack_file(filename);
let (_tmp_dir, tmp) = pack_test_tmp();
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("ref-delta-sha1.pack", HashKind::Sha1);
run_decode_with_ref_delta("ref-delta-sha256.pack", HashKind::Sha256);
}
fn run_decode_no_mem_limit(filename: &str, kind: HashKind) {
let _guard = set_hash_kind_for_test(kind);
let (source, _dl_guard) = download_pack_file(filename);
let (_tmp_dir, tmp) = pack_test_tmp();
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("small-sha1.pack", HashKind::Sha1);
run_decode_no_mem_limit("small-sha256.pack", HashKind::Sha256);
}
async fn run_decode_large_with_delta(filename: &str, kind: HashKind) {
let _guard = set_hash_kind_for_test(kind);
init_logger();
let (source, _dl_guard) = download_pack_file(filename);
let (_tmp_dir, tmp) = pack_test_tmp();
let f = fs::File::open(source).unwrap();
let mut buffered = BufReader::new(f);
let mut p = Pack::new(
Some(4),
Some(LARGE_PACK_TEST_MEM_LIMIT),
Some(tmp.clone()),
true,
);
let rt = p.decode(
&mut buffered,
|_obj| {
},
None::<fn(ObjectHash)>,
);
if let Err(e) = rt {
let _ = fs::remove_dir_all(&tmp);
panic!("Error: {e:?}");
}
}
#[cfg_attr(coverage, ignore)]
#[ignore = "requires large remote pack fixture"]
#[tokio::test]
async fn test_pack_decode_with_large_file_with_delta_without_ref() {
run_decode_large_with_delta("medium-sha1.pack", HashKind::Sha1).await;
run_decode_large_with_delta("medium-sha256.pack", HashKind::Sha256).await;
}
async fn run_decode_large_stream(filename: &str, kind: HashKind) {
let _guard = set_hash_kind_for_test(kind);
init_logger();
let (source, _dl_guard) = download_pack_file(filename);
let (_tmp_dir, tmp) = pack_test_tmp();
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(LARGE_PACK_TEST_MEM_LIMIT),
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();
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);
}
#[cfg_attr(coverage, ignore)]
#[ignore = "requires large remote pack fixture"]
#[tokio::test]
async fn test_decode_large_file_stream() {
run_decode_large_stream("medium-sha1.pack", HashKind::Sha1).await;
run_decode_large_stream("medium-sha256.pack", HashKind::Sha256).await;
}
async fn run_decode_large_file_async(filename: &str, kind: HashKind) {
let _guard = set_hash_kind_for_test(kind);
let (source, _dl_guard) = download_pack_file(filename);
let (_tmp_dir, tmp) = pack_test_tmp();
let f = fs::File::open(source).unwrap();
let buffered = BufReader::new(f);
let p = Pack::new(
Some(4),
Some(LARGE_PACK_TEST_MEM_LIMIT),
Some(tmp.clone()),
true,
);
let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel();
let handle = p.decode_async(buffered, tx); let mut cnt = 0;
while let Some(_entry) = rx.recv().await {
cnt += 1; }
let p = handle.join().unwrap();
assert_eq!(cnt, p.number);
}
#[cfg_attr(coverage, ignore)]
#[ignore = "requires large remote pack fixture"]
#[tokio::test]
async fn test_decode_large_file_async() {
run_decode_large_file_async("medium-sha1.pack", HashKind::Sha1).await;
run_decode_large_file_async("medium-sha256.pack", HashKind::Sha256).await;
}
fn run_decode_with_delta_no_ref(filename: &str, kind: HashKind) {
let _guard = set_hash_kind_for_test(kind);
let (source, _dl_guard) = download_pack_file(filename);
let (_tmp_dir, tmp) = pack_test_tmp();
let f = fs::File::open(source).unwrap();
let mut buffered = BufReader::new(f);
let mut p = Pack::new(None, Some(LARGE_PACK_TEST_MEM_LIMIT), Some(tmp), true);
p.decode(&mut buffered, |_| {}, None::<fn(ObjectHash)>)
.unwrap();
}
#[cfg_attr(coverage, ignore)]
#[ignore = "requires large remote pack fixture"]
#[test]
fn test_pack_decode_with_delta_without_ref() {
run_decode_with_delta_no_ref("medium-sha1.pack", HashKind::Sha1);
run_decode_with_delta_no_ref("medium-sha256.pack", HashKind::Sha256);
}
#[cfg_attr(coverage, ignore)]
#[ignore = "requires large remote pack fixture"]
#[test] 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 (kind, filename) in [
(HashKind::Sha1, "medium-sha1.pack"),
(HashKind::Sha256, "medium-sha256.pack"),
] {
let f1 = run_decode_large_with_delta(filename, kind);
let f2 = run_decode_large_with_delta(filename, kind);
let _ = futures::future::join(f1, f2).await;
}
});
}
#[test]
fn test_stats_pack_small_sha1() {
let _guard = set_hash_kind_for_test(HashKind::Sha1);
let (source, _dl_guard) = download_pack_file("small-sha1.pack");
let stats = Pack::stats_pack(source).expect("stats_pack should succeed");
eprintln!(
"small-sha1 stats: total={}, commits={}, trees={}, blobs={}, tags={}, deltas={}",
stats.total, stats.commits, stats.trees, stats.blobs, stats.tags, stats.deltas
);
let sum = stats.commits + stats.trees + stats.blobs + stats.tags + stats.deltas;
assert_eq!(
sum, stats.total,
"per-type counts should sum to total ({} vs {})",
sum, stats.total
);
assert!(stats.commits > 0, "expected at least one commit");
assert!(stats.blobs > 0, "expected at least one blob");
}
#[test]
fn test_stats_pack_medium_sha1_has_deltas() {
let _guard = set_hash_kind_for_test(HashKind::Sha1);
let (source, _dl_guard) = download_pack_file("medium-sha1.pack");
let stats = Pack::stats_pack(source).expect("stats_pack should succeed on medium pack");
eprintln!(
"medium-sha1 stats: total={}, commits={}, trees={}, blobs={}, tags={}, deltas={}",
stats.total, stats.commits, stats.trees, stats.blobs, stats.tags, stats.deltas
);
let sum = stats.commits + stats.trees + stats.blobs + stats.tags + stats.deltas;
assert_eq!(sum, stats.total, "per-type counts must equal total");
assert!(
stats.deltas > 0,
"expected delta objects in medium-sha1 pack"
);
assert!(stats.total > 1000, "expected a sizeable medium pack");
}
#[test]
fn test_stats_pack_file_not_found() {
let result = Pack::stats_pack(PathBuf::from("/nonexistent/path/to/fake.pack"));
assert!(
result.is_err(),
"stats_pack should return Err for a missing file"
);
}
#[test]
fn test_stats_pack_invalid_pack_magic() {
use std::io::Write;
use tempfile::NamedTempFile;
let mut tmp = NamedTempFile::new().expect("create temp file");
tmp.write_all(b"FAKE\x00\x00\x00\x02\x00\x00\x00\x05")
.expect("write temp bytes");
let path = tmp.path().to_path_buf();
let result = Pack::stats_pack(path);
assert!(
result.is_err(),
"stats_pack should return Err for invalid pack magic"
);
}
}