use crate::zlib::stream::inflate::InflateReaderBoxed; use crate::{hash, pack, zlib::stream::inflate::Inflate}; use git_features::hash::Sha1; use git_object::owned; use quick_error::quick_error; use std::{fs, io}; quick_error! { #[derive(Debug)] pub enum Error { Io(err: io::Error) { display("An IO operation failed while streaming an entry") from() source(err) } PackParse(err: pack::data::parse::Error) { display("The pack header could not be parsed") from() source(err) } ChecksumMismatch { expected: owned::Id, actual: owned::Id } { display("pack checksum in trailer was {}, but actual checksum was {}", expected, actual) } } } #[derive(PartialEq, Eq, Debug, Hash, Ord, PartialOrd, Clone)] #[cfg_attr(feature = "serde1", derive(serde::Serialize, serde::Deserialize))] pub struct Entry { pub header: pack::data::Header, /// amount of bytes used to encode the `header`. `pack_offset + header_size` is the beginning of the compressed data in the pack. pub header_size: u16, pub pack_offset: u64, /// The bytes consumed while producing `decompressed` /// These do not contain the header, which makes it possible to easily replace a RefDelta with offset deltas /// when resolving thin packs. /// Depends on `CompressionMode` when the iterator is initialized pub compressed: Option>, /// The amount of bytes the compressed portion of the entry takes. pub compressed_size: u64, /// The CRC32 over the complete entry, that is encoded header and compressed object data. /// Depends on `CompressionMode` when the iterator is initialized pub crc32: Option, /// The amount of decompressed bytes pub decompressed_size: u64, /// Set for the last object in the iteration, providing the hash over all bytes of the iteration /// for use as trailer in a pack pub trailer: Option, } pub struct Iter { read: R, decompressor: Option>, offset: u64, had_error: bool, kind: pack::data::Kind, objects_left: u32, hash: Option, mode: Mode, compressed: CompressedBytesMode, compressed_buf: Option>, } #[derive(PartialEq, Eq, Debug, Hash, Ord, PartialOrd, Clone, Copy)] #[cfg_attr(feature = "serde1", derive(serde::Serialize, serde::Deserialize))] pub enum Mode { /// Provide the trailer as read from the pack AsIs, /// Generate an own hash and trigger an error on the last iterated object /// if it does not match the hash provided with the pack Verify, /// Generate an own hash and if there was an error or the objects are depleted early /// due to partial packs, return the last valid entry and with our own hash thus far. /// Note that the existing pack hash, if present, will be ignored. /// As we won't know which objects fails, every object will have the hash obtained thus far. /// This also means that algorithms must know about this possibility, or else might wrongfully /// assume the pack is finished. Restore, } #[derive(PartialEq, Eq, Debug, Hash, Ord, PartialOrd, Clone, Copy)] #[cfg_attr(feature = "serde1", derive(serde::Serialize, serde::Deserialize))] pub enum CompressedBytesMode { /// Do nothing with the compressed bytes we read Ignore, /// Only create a CRC32 of the entry, otherwise similar to `Ignore` CRC32, /// Keep them and pass them along in a newly allocated buffer Keep, /// As above, but also compute a CRC32 KeepAndCRC32, } impl CompressedBytesMode { pub fn crc32(&self) -> bool { match self { CompressedBytesMode::KeepAndCRC32 | CompressedBytesMode::CRC32 => true, CompressedBytesMode::Keep | CompressedBytesMode::Ignore => false, } } pub fn keep(&self) -> bool { match self { CompressedBytesMode::Keep | CompressedBytesMode::KeepAndCRC32 => true, CompressedBytesMode::Ignore | CompressedBytesMode::CRC32 => false, } } } impl Iter where R: io::BufRead, { pub fn kind(&self) -> pack::data::Kind { self.kind } pub fn mode(&self) -> Mode { self.mode } /// Note that `read` is expected at the beginning of a valid pack file with header and trailer /// If `verify` is true, we will assert the SHA1 is actually correct before returning the last entry. /// Otherwise bit there is a chance that some kinds of bitrot or inconsistencies will not be detected. pub fn new_from_header(mut read: R, trailer: Mode, compressed: CompressedBytesMode) -> Result, Error> { let mut header_data = [0u8; 12]; read.read_exact(&mut header_data)?; let (kind, num_objects) = pack::data::parse::header(&header_data)?; assert_eq!( kind, pack::data::Kind::V2, "let's stop here if we see undocumented pack formats" ); Ok(Iter { read, decompressor: None, compressed, offset: 12, had_error: false, kind, objects_left: num_objects, hash: if trailer != Mode::AsIs { let mut hash = Sha1::default(); hash.update(&header_data); Some(hash) } else { None }, mode: trailer, compressed_buf: None, }) } fn next_inner(&mut self) -> Result { self.objects_left -= 1; // even an error counts as objects // Read header let entry = match self.hash.take() { Some(hash) => { let mut read = read_and_pass_to( &mut self.read, hash::Write { inner: io::sink(), hash, }, ); let res = pack::data::Entry::from_read(&mut read, self.offset); self.hash = Some(read.write.hash); res } None => pack::data::Entry::from_read(&mut self.read, self.offset), } .map_err(Error::from)?; // Decompress object to learn it's compressed bytes let mut decompressor = self.decompressor.take().unwrap_or_default(); let compressed_buf = self.compressed_buf.take().unwrap_or_else(|| Vec::with_capacity(4096)); decompressor.reset(); let mut decompressed_reader = InflateReaderBoxed { inner: read_and_pass_to( &mut self.read, if self.compressed.keep() { Vec::with_capacity(entry.decompressed_size as usize) } else { compressed_buf }, ), decompressor, }; let bytes_copied = io::copy(&mut decompressed_reader, &mut io::sink())?; debug_assert_eq!( bytes_copied, entry.decompressed_size, "We should have decompressed {} bytes, but got {} instead", entry.decompressed_size, bytes_copied ); let pack_offset = self.offset; let compressed_size = decompressed_reader.decompressor.total_in; self.offset += entry.header_size() as u64 + compressed_size; self.decompressor = Some(decompressed_reader.decompressor); let mut compressed = decompressed_reader.inner.write; debug_assert_eq!( compressed_size, compressed.len() as u64, "we must track exactly the same amount of bytes as read by the decompressor" ); if let Some(hash) = self.hash.as_mut() { hash.update(&compressed); } let crc32 = if self.compressed.crc32() { let mut header_buf = [0u8; 32]; let header_len = entry.header.to_write(bytes_copied, header_buf.as_mut())?; let state = git_features::hash::crc32_update(0, &header_buf[..header_len]); Some(git_features::hash::crc32_update(state, &compressed)) } else { None }; let compressed = if self.compressed.keep() { Some(compressed) } else { compressed.clear(); self.compressed_buf = Some(compressed); None }; // Last objects gets trailer (which is potentially verified) let trailer = if self.objects_left == 0 { let mut id = owned::Id::from([0; 20]); if let Err(err) = self.read.read_exact(id.as_mut_slice()) { if self.mode != Mode::Restore { return Err(err.into()); } } if let Some(hash) = self.hash.take() { let actual_id = owned::Id::from(hash.digest()); if self.mode == Mode::Restore { id = actual_id; } if id != actual_id { return Err(Error::ChecksumMismatch { actual: actual_id, expected: id, }); } } Some(id) } else if self.mode == Mode::Restore { let hash = self.hash.clone().expect("in restore mode a hash is set"); Some(owned::Id::from(hash.digest())) } else { None }; Ok(Entry { header: entry.header, header_size: entry.header_size() as u16, compressed, compressed_size, crc32, pack_offset, decompressed_size: bytes_copied, trailer, }) } } fn read_and_pass_to(read: &mut R, to: W) -> PassThrough<&mut R, W> { PassThrough { read, write: to } } impl Iterator for Iter where R: io::BufRead, { type Item = Result; fn next(&mut self) -> Option { if self.had_error || self.objects_left == 0 { return None; } let result = self.next_inner(); self.had_error = result.is_err(); if self.had_error { self.objects_left = 0; } if self.mode == Mode::Restore && self.had_error { None } else { Some(result) } } fn size_hint(&self) -> (usize, Option) { (self.objects_left as usize, Some(self.objects_left as usize)) } } impl std::iter::ExactSizeIterator for Iter where R: io::BufRead {} struct PassThrough { read: R, write: W, } impl io::BufRead for PassThrough where Self: io::Read, R: io::BufRead, W: io::Write, { fn fill_buf(&mut self) -> io::Result<&[u8]> { self.read.fill_buf() } fn consume(&mut self, amt: usize) { let buf = self .read .fill_buf() .expect("never fail as we called fill-buf before and this does nothing"); self.write .write_all(&buf[..amt]) .expect("a write to never fail - should be a memory buffer"); self.read.consume(amt) } } impl io::Read for PassThrough where W: io::Write, R: io::Read, { fn read(&mut self, buf: &mut [u8]) -> io::Result { let bytes_read = self.read.read(buf)?; self.write.write_all(&buf[..bytes_read])?; Ok(bytes_read) } } impl pack::data::File { /// Returns an iterator over the pack file itself, without making use of the memory mapping. /// /// Note that this iterator is costly as no pack index is used, forcing each entry to be decompressed. /// If an index is available, use the `traverse(…)` method instead for maximum performance. pub fn streaming_iter(&self) -> Result, Error> { let reader = io::BufReader::with_capacity(4096 * 8, fs::File::open(&self.path)?); Iter::new_from_header(reader, Mode::Verify, CompressedBytesMode::KeepAndCRC32) } }