use crate::delta;
use crate::hash::{self, Hash};
use crate::object::{MkitError, Object};
use crate::store::{MAX_RAW_OBJECT_SIZE, ObjectStore};
use std::borrow::Cow;
use std::sync::atomic::{AtomicU64, Ordering};
pub const MAGIC: &[u8; 4] = b"MKIT";
pub const VERSION: u32 = 1;
pub const VERSION_V2: u32 = 2;
pub const MAX_ENTRIES: u32 = 10_000_000;
pub const MAX_TOTAL_PAYLOAD: u64 = 4 * 1024 * 1024 * 1024;
pub const TRAILER_LEN: usize = 32;
pub const HEADER_LEN: usize = 4 + 4 + 4;
pub const ENTRY_FRAME_LEN: usize = 1 + 4;
pub const VERSION_OFFSET: usize = 4;
pub const ENTRY_COUNT_OFFSET: usize = 8;
#[cfg(feature = "pack-zstd")]
const MIN_COMPRESS_LEN: usize = 64;
#[cfg(feature = "pack-zstd")]
const ZSTD_LEVEL: i32 = 3;
const ZSTD_LEN_PREFIX: usize = 4;
#[derive(Debug, thiserror::Error)]
pub enum PackError {
#[error("packfile is shorter than the {HEADER_LEN}-byte header + {TRAILER_LEN}-byte trailer")]
PackfileTooShort,
#[error("first 4 bytes are not ASCII \"MKIT\"")]
InvalidMagic,
#[error("version {0} is not supported (v1 or v2 only)")]
UnsupportedVersion(u32),
#[error(
"entry_type {0:#04x} is not 0x00 (raw), 0x02 (delta), 0x03 (zstd-raw), or 0x04 \
(zstd-delta) — or is a v2-only entry type inside a version-1 pack"
)]
InvalidEntryType(u8),
#[error("entry_count {0} exceeds the {MAX_ENTRIES} cap")]
TooManyObjects(u32),
#[error("sum of payload_len exceeds {MAX_TOTAL_PAYLOAD} bytes")]
PackfileTooLarge,
#[error("entry payload extends past the trailer offset")]
UnexpectedEof,
#[error("trailer BLAKE3 mismatch — packfile is corrupt or truncated")]
PackfileCorrupted,
#[error("delta entry references base hash {0} which is not in this pack or the store")]
DeltaBaseMissing(String),
#[error("delta entry payload is shorter than the 32-byte base hash prefix")]
DeltaEntryTruncated,
#[error("delta reconstruction failed: {0}")]
DeltaApply(#[from] MkitError),
#[error("pack entry is not a canonical storable object: {0}")]
InvalidObject(MkitError),
#[error("pack entry resolves to pack-only delta object")]
NonStorableObject,
#[error("pack contains trailing bytes after declared entries")]
TrailingData,
#[error("store I/O failure: {0}")]
Store(#[from] crate::store::StoreError),
#[error("zstd entry payload is shorter than its length-prefix header")]
ZstdEntryTruncated,
#[error(
"zstd entry's claimed decompressed size {0} exceeds the {MAX_RAW_OBJECT_SIZE}-byte cap"
)]
DecompressedSizeOverCap(usize),
#[error("zstd entry claims {0} decompressed bytes but produced {1}")]
DecompressedSizeMismatch(usize, usize),
#[error("zstd decompression failed: {0}")]
ZstdDecompress(String),
}
#[derive(Debug, Clone, Default, PartialEq, Eq)]
pub struct UnpackReport {
pub raw_count: u32,
pub delta_count: u32,
pub stored: Vec<Hash>,
}
#[derive(Debug)]
pub struct PackWriter {
buf: Vec<u8>,
entry_count: u32,
total_payload: u64,
has_compressed_entry: bool,
}
impl Default for PackWriter {
fn default() -> Self {
Self::new()
}
}
impl PackWriter {
#[must_use]
pub fn new() -> Self {
let mut buf = Vec::with_capacity(HEADER_LEN);
buf.extend_from_slice(MAGIC);
buf.extend_from_slice(&VERSION.to_le_bytes());
buf.extend_from_slice(&0u32.to_le_bytes()); Self {
buf,
entry_count: 0,
total_payload: 0,
has_compressed_entry: false,
}
}
pub fn push_raw(&mut self, hash_of_bytes: Hash, bytes: &[u8]) -> Result<Hash, PackError> {
if let Some(frame) = maybe_compress(bytes) {
let uncompressed_len: u32 = bytes
.len()
.try_into()
.map_err(|_| PackError::PackfileTooLarge)?;
let payload_len = ZSTD_LEN_PREFIX + frame.len();
self.check_caps_for(payload_len)?;
self.total_payload += payload_len as u64;
self.append_entry(0x03, &[&uncompressed_len.to_le_bytes(), &frame])?;
self.has_compressed_entry = true;
} else {
self.check_caps_for(bytes.len())?;
self.total_payload += bytes.len() as u64;
self.append_entry(0x00, &[bytes])?;
}
self.entry_count += 1;
Ok(hash_of_bytes)
}
pub fn push_delta(&mut self, base_hash: &Hash, delta_stream: &[u8]) -> Result<(), PackError> {
if let Some(frame) = maybe_compress(delta_stream) {
let uncompressed_len: u32 = delta_stream
.len()
.try_into()
.map_err(|_| PackError::PackfileTooLarge)?;
let payload_len = hash::HASH_LEN + ZSTD_LEN_PREFIX + frame.len();
self.check_caps_for(payload_len)?;
self.total_payload += payload_len as u64;
self.append_entry(
0x04,
&[
base_hash.as_slice(),
&uncompressed_len.to_le_bytes(),
&frame,
],
)?;
self.has_compressed_entry = true;
} else {
let payload_len = hash::HASH_LEN + delta_stream.len();
self.check_caps_for(payload_len)?;
self.total_payload += payload_len as u64;
self.append_entry(0x02, &[base_hash.as_slice(), delta_stream])?;
}
self.entry_count += 1;
Ok(())
}
fn append_entry(&mut self, etype: u8, parts: &[&[u8]]) -> Result<(), PackError> {
let payload_len: usize = parts.iter().map(|p| p.len()).sum();
let plen: u32 = payload_len
.try_into()
.map_err(|_| PackError::PackfileTooLarge)?;
self.buf.push(etype);
self.buf.extend_from_slice(&plen.to_le_bytes());
for p in parts {
self.buf.extend_from_slice(p);
}
Ok(())
}
fn check_caps_for(&self, add_len: usize) -> Result<(), PackError> {
let next_count = u64::from(self.entry_count) + 1;
if next_count > u64::from(MAX_ENTRIES) {
return Err(PackError::TooManyObjects(MAX_ENTRIES + 1));
}
let next_total = self.total_payload.saturating_add(add_len as u64);
if next_total > MAX_TOTAL_PAYLOAD {
return Err(PackError::PackfileTooLarge);
}
Ok(())
}
#[must_use]
pub fn entry_count(&self) -> usize {
self.entry_count as usize
}
#[must_use]
pub fn total_payload(&self) -> u64 {
self.total_payload
}
pub fn finish(self) -> Result<Vec<u8>, PackError> {
self.finish_inner(None)
}
#[cfg(test)]
pub(crate) fn finish_tracking_bytes_copied(
self,
bytes_copied: &AtomicU64,
) -> Result<Vec<u8>, PackError> {
self.finish_inner(Some(bytes_copied))
}
fn finish_inner(mut self, bytes_copied: Option<&AtomicU64>) -> Result<Vec<u8>, PackError> {
if self.entry_count > MAX_ENTRIES {
return Err(PackError::TooManyObjects(self.entry_count));
}
let version = if self.has_compressed_entry {
VERSION_V2
} else {
VERSION
};
self.buf[VERSION_OFFSET..VERSION_OFFSET + 4].copy_from_slice(&version.to_le_bytes());
self.buf[ENTRY_COUNT_OFFSET..ENTRY_COUNT_OFFSET + 4]
.copy_from_slice(&self.entry_count.to_le_bytes());
let trailer = hash::hash(&self.buf);
if let Some(c) = bytes_copied {
c.fetch_add(trailer.len() as u64, Ordering::Relaxed);
}
self.buf.extend_from_slice(&trailer);
Ok(self.buf)
}
}
#[must_use]
pub fn pack_key(pack_bytes: &[u8]) -> Hash {
hash::hash(pack_bytes)
}
#[cfg(feature = "pack-zstd")]
fn maybe_compress(data: &[u8]) -> Option<Vec<u8>> {
if data.len() < MIN_COMPRESS_LEN {
return None;
}
let compressed = zstd::bulk::compress(data, ZSTD_LEVEL).ok()?;
if ZSTD_LEN_PREFIX + compressed.len() < data.len() {
Some(compressed)
} else {
None
}
}
#[cfg(not(feature = "pack-zstd"))]
fn maybe_compress(_data: &[u8]) -> Option<Vec<u8>> {
None
}
fn decompress_zstd_entry(payload: &[u8]) -> Result<Vec<u8>, PackError> {
if payload.len() < ZSTD_LEN_PREFIX {
return Err(PackError::ZstdEntryTruncated);
}
let uncompressed_len =
u32::from_le_bytes(payload[..ZSTD_LEN_PREFIX].try_into().expect("4 bytes")) as usize;
if uncompressed_len > MAX_RAW_OBJECT_SIZE {
return Err(PackError::DecompressedSizeOverCap(uncompressed_len));
}
let frame = &payload[ZSTD_LEN_PREFIX..];
let decompressed = zstd_decompress_capped(frame, uncompressed_len)?;
if decompressed.len() != uncompressed_len {
return Err(PackError::DecompressedSizeMismatch(
uncompressed_len,
decompressed.len(),
));
}
Ok(decompressed)
}
#[cfg(feature = "pack-zstd")]
fn zstd_decompress_capped(frame: &[u8], capacity: usize) -> Result<Vec<u8>, PackError> {
zstd::bulk::decompress(frame, capacity).map_err(|e| PackError::ZstdDecompress(e.to_string()))
}
#[cfg(not(feature = "pack-zstd"))]
fn zstd_decompress_capped(_frame: &[u8], _capacity: usize) -> Result<Vec<u8>, PackError> {
Err(PackError::ZstdDecompress(
"this build was compiled without the `pack-zstd` feature".to_string(),
))
}
pub fn delta_base_hashes(pack_bytes: &[u8]) -> Result<Vec<Hash>, PackError> {
if pack_bytes.len() < HEADER_LEN + TRAILER_LEN {
return Err(PackError::PackfileTooShort);
}
if &pack_bytes[..4] != MAGIC.as_slice() {
return Err(PackError::InvalidMagic);
}
let version = u32::from_le_bytes(pack_bytes[4..8].try_into().expect("4 bytes"));
if version != VERSION && version != VERSION_V2 {
return Err(PackError::UnsupportedVersion(version));
}
let count = u32::from_le_bytes(
pack_bytes[ENTRY_COUNT_OFFSET..ENTRY_COUNT_OFFSET + 4]
.try_into()
.expect("4 bytes"),
);
if count > MAX_ENTRIES {
return Err(PackError::TooManyObjects(count));
}
let split = pack_bytes.len() - TRAILER_LEN;
let mut bases = Vec::new();
let mut seen = std::collections::HashSet::new();
let mut pos = HEADER_LEN;
for _ in 0..count {
if pos + ENTRY_FRAME_LEN > split {
return Err(PackError::UnexpectedEof);
}
let etype = pack_bytes[pos];
pos += 1;
let payload_len =
u32::from_le_bytes(pack_bytes[pos..pos + 4].try_into().expect("4 bytes")) as usize;
pos += 4;
if pos + payload_len > split {
return Err(PackError::UnexpectedEof);
}
if etype == 0x02 || etype == 0x04 {
if payload_len < TRAILER_LEN {
return Err(PackError::DeltaEntryTruncated);
}
let mut base = [0u8; 32];
base.copy_from_slice(&pack_bytes[pos..pos + TRAILER_LEN]);
if seen.insert(base) {
bases.push(base);
}
}
pos += payload_len;
}
Ok(bases)
}
#[derive(Debug)]
pub struct PackReader;
impl PackReader {
pub fn read(pack_bytes: &[u8], store: &ObjectStore) -> Result<UnpackReport, PackError> {
Self::read_with_payload_cap(pack_bytes, store, MAX_TOTAL_PAYLOAD)
}
pub(crate) fn read_with_payload_cap(
pack_bytes: &[u8],
store: &ObjectStore,
payload_cap: u64,
) -> Result<UnpackReport, PackError> {
Self::read_inner(pack_bytes, store, payload_cap, None)
}
#[cfg(test)]
pub(crate) fn read_tracking_owned_bytes(
pack_bytes: &[u8],
store: &ObjectStore,
owned_bytes: &AtomicU64,
) -> Result<UnpackReport, PackError> {
Self::read_inner(pack_bytes, store, MAX_TOTAL_PAYLOAD, Some(owned_bytes))
}
fn read_inner(
pack_bytes: &[u8],
store: &ObjectStore,
payload_cap: u64,
owned_bytes: Option<&AtomicU64>,
) -> Result<UnpackReport, PackError> {
let (version, split, count) = validate_pack_header(pack_bytes)?;
let mut report = UnpackReport::default();
let mut in_pack: std::collections::HashMap<Hash, Cow<'_, [u8]>> =
std::collections::HashMap::new();
let mut total_payload: u64 = 0;
let mut pos = HEADER_LEN;
let batch = store.batch();
for _ in 0..count {
if pos + ENTRY_FRAME_LEN > split {
return Err(PackError::UnexpectedEof);
}
let etype = pack_bytes[pos];
pos += 1;
let payload_len =
u32::from_le_bytes(pack_bytes[pos..pos + 4].try_into().expect("4 bytes")) as usize;
pos += 4;
total_payload = total_payload.saturating_add(payload_len as u64);
if total_payload > payload_cap {
return Err(PackError::PackfileTooLarge);
}
if pos + payload_len > split {
return Err(PackError::UnexpectedEof);
}
let payload = &pack_bytes[pos..pos + payload_len];
pos += payload_len;
match etype {
0x00 => {
stage_raw_object(
&batch,
&mut in_pack,
&mut report,
owned_bytes,
Cow::Borrowed(payload),
)?;
}
0x02 => {
if payload.len() < hash::HASH_LEN {
return Err(PackError::DeltaEntryTruncated);
}
let mut base_hash = [0u8; hash::HASH_LEN];
base_hash.copy_from_slice(&payload[..hash::HASH_LEN]);
let stream = &payload[hash::HASH_LEN..];
stage_delta_target(
store,
&batch,
&mut in_pack,
&mut report,
owned_bytes,
base_hash,
stream,
)?;
}
0x03 if version == VERSION_V2 => {
let obj_bytes = decompress_zstd_entry(payload)?;
stage_raw_object(
&batch,
&mut in_pack,
&mut report,
owned_bytes,
Cow::Owned(obj_bytes),
)?;
}
0x04 if version == VERSION_V2 => {
if payload.len() < hash::HASH_LEN {
return Err(PackError::DeltaEntryTruncated);
}
let mut base_hash = [0u8; hash::HASH_LEN];
base_hash.copy_from_slice(&payload[..hash::HASH_LEN]);
let stream = decompress_zstd_entry(&payload[hash::HASH_LEN..])?;
stage_delta_target(
store,
&batch,
&mut in_pack,
&mut report,
owned_bytes,
base_hash,
&stream,
)?;
}
0x01 => return Err(PackError::InvalidEntryType(0x01)),
other => return Err(PackError::InvalidEntryType(other)),
}
}
if pos != split {
return Err(PackError::TrailingData);
}
batch.commit()?;
Ok(report)
}
}
fn validate_pack_header(pack_bytes: &[u8]) -> Result<(u32, usize, u32), PackError> {
if pack_bytes.len() < HEADER_LEN + TRAILER_LEN {
return Err(PackError::PackfileTooShort);
}
if &pack_bytes[..4] != MAGIC.as_slice() {
return Err(PackError::InvalidMagic);
}
let version = u32::from_le_bytes(pack_bytes[4..8].try_into().expect("4 bytes"));
if version != VERSION && version != VERSION_V2 {
return Err(PackError::UnsupportedVersion(version));
}
let split = pack_bytes.len() - TRAILER_LEN;
let body = &pack_bytes[..split];
let trailer = &pack_bytes[split..];
let computed = hash::hash(body);
if computed.as_slice() != trailer {
return Err(PackError::PackfileCorrupted);
}
let count = u32::from_le_bytes(
pack_bytes[ENTRY_COUNT_OFFSET..ENTRY_COUNT_OFFSET + 4]
.try_into()
.expect("4 bytes"),
);
if count > MAX_ENTRIES {
return Err(PackError::TooManyObjects(count));
}
let body_after_header = body.len() - HEADER_LEN;
if u64::from(count) * ENTRY_FRAME_LEN as u64 > body_after_header as u64 {
return Err(PackError::TooManyObjects(count));
}
Ok((version, split, count))
}
fn stage_raw_object<'p>(
batch: &crate::batch::WriteBatch<'_>,
in_pack: &mut std::collections::HashMap<Hash, Cow<'p, [u8]>>,
report: &mut UnpackReport,
owned_bytes: Option<&AtomicU64>,
payload: Cow<'p, [u8]>,
) -> Result<(), PackError> {
let obj = validate_storable_object(&payload)?;
let stored_hash = crate::object::id_from_object(&obj, &payload);
batch.write_prehashed(stored_hash, &[payload.as_ref()])?;
if let (Cow::Owned(_), Some(c)) = (&payload, owned_bytes) {
c.fetch_add(payload.len() as u64, Ordering::Relaxed);
}
in_pack.insert(stored_hash, payload);
report.raw_count += 1;
report.stored.push(stored_hash);
Ok(())
}
#[allow(clippy::too_many_arguments)]
fn stage_delta_target(
store: &ObjectStore,
batch: &crate::batch::WriteBatch<'_>,
in_pack: &mut std::collections::HashMap<Hash, Cow<'_, [u8]>>,
report: &mut UnpackReport,
owned_bytes: Option<&AtomicU64>,
base_hash: Hash,
stream: &[u8],
) -> Result<(), PackError> {
let resolved = resolve_delta_target(store, in_pack, base_hash, stream)?;
let obj = validate_storable_object(&resolved)?;
let stored_hash = crate::object::id_from_object(&obj, &resolved);
batch.write_prehashed(stored_hash, &[&resolved])?;
if let Some(c) = owned_bytes {
c.fetch_add(resolved.len() as u64, Ordering::Relaxed);
}
in_pack.insert(stored_hash, Cow::Owned(resolved));
report.delta_count += 1;
report.stored.push(stored_hash);
Ok(())
}
fn resolve_delta_target(
store: &ObjectStore,
in_pack: &mut std::collections::HashMap<Hash, Cow<'_, [u8]>>,
base_hash: Hash,
stream: &[u8],
) -> Result<Vec<u8>, PackError> {
let base_bytes: Cow<'_, [u8]> = if let Some(b) = in_pack.get(&base_hash) {
Cow::Borrowed(b.as_ref())
} else if store.contains(&base_hash) {
let bytes = store.read(&base_hash)?;
validate_storable_object(&bytes)?;
in_pack.insert(base_hash, Cow::Owned(bytes.clone()));
Cow::Owned(bytes)
} else {
return Err(PackError::DeltaBaseMissing(hash::to_hex(&base_hash)));
};
validate_delta_result_size(stream)?;
let resolved = delta::decode(base_bytes.as_ref(), stream)?;
Ok(resolved)
}
fn validate_storable_object(bytes: &[u8]) -> Result<Object, PackError> {
if bytes.len() > MAX_RAW_OBJECT_SIZE {
return Err(PackError::Store(crate::store::StoreError::ObjectTooLarge));
}
match crate::serialize::deserialize(bytes).map_err(PackError::InvalidObject)? {
Object::Delta(_) => Err(PackError::NonStorableObject),
obj @ (Object::Blob(_)
| Object::Tree(_)
| Object::Commit(_)
| Object::Remix(_)
| Object::ChunkedBlob(_)
| Object::Tag(_)) => Ok(obj),
}
}
fn validate_delta_result_size(stream: &[u8]) -> Result<(), PackError> {
if stream.len() < delta::HEADER_LEN {
return Err(PackError::DeltaApply(MkitError::UnexpectedEof));
}
let result_len = u32::from_le_bytes(stream[5..9].try_into().expect("4 bytes")) as usize;
if result_len > MAX_RAW_OBJECT_SIZE {
return Err(PackError::Store(crate::store::StoreError::ObjectTooLarge));
}
Ok(())
}
#[cfg(test)]
mod tests {
use super::*;
use tempfile::TempDir;
fn fresh_store() -> (TempDir, ObjectStore) {
let dir = TempDir::new().unwrap();
let store = ObjectStore::init(&crate::layout::RepoLayout::single(dir.path())).unwrap();
(dir, store)
}
fn write_blob_via_serialize(payload: &[u8]) -> Vec<u8> {
let blob = crate::object::Object::Blob(crate::object::Blob {
data: payload.to_vec(),
});
crate::serialize::serialize(&blob).expect("serialize blob")
}
fn finish_pack_body(mut body: Vec<u8>) -> Vec<u8> {
let trailer = hash::hash(&body);
body.extend_from_slice(&trailer);
body
}
fn incompressible_bytes(seed: u64, len: usize) -> Vec<u8> {
let mut buf = vec![0u8; len];
let mut state = seed | 1; for chunk in buf.chunks_mut(8) {
state = state
.wrapping_mul(6_364_136_223_846_793_005)
.wrapping_add(1_442_695_040_888_963_407);
let bytes = state.to_le_bytes();
chunk.copy_from_slice(&bytes[..chunk.len()]);
}
buf
}
#[test]
fn empty_pack_is_44_bytes() {
let pack = PackWriter::new().finish().unwrap();
assert_eq!(pack.len(), HEADER_LEN + TRAILER_LEN);
assert_eq!(&pack[..4], MAGIC);
assert_eq!(u32::from_le_bytes(pack[4..8].try_into().unwrap()), VERSION);
assert_eq!(
u32::from_le_bytes(
pack[ENTRY_COUNT_OFFSET..ENTRY_COUNT_OFFSET + 4]
.try_into()
.unwrap()
),
0
);
let (_dir, store) = fresh_store();
let report = PackReader::read(&pack, &store).unwrap();
assert_eq!(report.raw_count, 0);
assert_eq!(report.delta_count, 0);
assert!(report.stored.is_empty());
}
#[test]
fn unpack_writes_objects_via_single_batch_flush() {
use crate::batch::testing::{Ev, RecordingSyncer};
use std::sync::Arc;
let mut w = PackWriter::new();
let mut blobs = Vec::new();
for i in 0u32..30 {
let blob = write_blob_via_serialize(format!("pack object {i}").as_bytes());
w.push_raw(hash::hash(&blob), &blob).unwrap();
blobs.push(blob);
}
let pack = w.finish().unwrap();
let (_dir, mut store) = fresh_store();
let rec = Arc::new(RecordingSyncer::default());
store.set_syncer(rec.clone());
let report = PackReader::read(&pack, &store).unwrap();
assert_eq!(report.raw_count, 30);
let fulls = rec
.events()
.iter()
.filter(|e| matches!(e, Ev::Full(_)))
.count();
assert_eq!(
fulls, 2,
"unpack flush cost must be constant, not O(objects)"
);
for blob in &blobs {
assert_eq!(store.read(&hash::hash(blob)).unwrap(), *blob);
}
}
#[test]
fn single_raw_roundtrip() {
let blob = write_blob_via_serialize(b"hello packfile");
let h = hash::hash(&blob);
let mut w = PackWriter::new();
w.push_raw(h, &blob).unwrap();
let pack = w.finish().unwrap();
let (_dir, store) = fresh_store();
let report = PackReader::read(&pack, &store).unwrap();
assert_eq!(report.raw_count, 1);
assert_eq!(report.delta_count, 0);
assert_eq!(report.stored, vec![h]);
assert_eq!(store.read(&h).unwrap(), blob);
}
#[test]
fn total_payload_tracks_wire_sum_for_mixed_raw_and_delta() {
let mut w = PackWriter::new();
assert_eq!(w.total_payload(), 0);
let raw = write_blob_via_serialize(&incompressible_bytes(0xA11C_E000, 2048));
let raw_hash = hash::hash(&raw);
w.push_raw(raw_hash, &raw).unwrap();
assert_eq!(w.total_payload(), raw.len() as u64);
let base = write_blob_via_serialize(&incompressible_bytes(0xB0BA_1000, 2048));
let base_hash = hash::hash(&base);
let target = write_blob_via_serialize(&incompressible_bytes(0xC0FF_EE00, 2048));
let stream = delta::encode(&base, &target).unwrap();
let before_delta = w.total_payload();
w.push_delta(&base_hash, &stream).unwrap();
let delta_wire_len = w.total_payload() - before_delta;
assert!(delta_wire_len <= (hash::HASH_LEN + stream.len()) as u64);
assert_eq!(w.total_payload(), before_delta + delta_wire_len);
assert!(w.total_payload() <= raw.len() as u64 + (hash::HASH_LEN + stream.len()) as u64);
}
#[test]
fn raw_then_delta_resolves_in_pack() {
let mut content_base = vec![0u8; 1024];
for (i, b) in content_base.iter_mut().enumerate() {
*b = u8::try_from(i % 251).expect("modulo < 256");
}
let mut content_target = content_base.clone();
content_target[500] = 0xFF;
content_target[501] = 0xFE;
let base_obj = write_blob_via_serialize(&content_base);
let target_obj = write_blob_via_serialize(&content_target);
let base_hash = hash::hash(&base_obj);
let target_hash = hash::hash(&target_obj);
let stream = delta::encode(&base_obj, &target_obj).unwrap();
let mut w = PackWriter::new();
w.push_raw(base_hash, &base_obj).unwrap();
w.push_delta(&base_hash, &stream).unwrap();
let pack = w.finish().unwrap();
let (_dir, store) = fresh_store();
let report = PackReader::read(&pack, &store).unwrap();
assert_eq!(report.raw_count, 1);
assert_eq!(report.delta_count, 1);
assert_eq!(report.stored, vec![base_hash, target_hash]);
assert_eq!(store.read(&target_hash).unwrap(), target_obj);
}
#[test]
fn delta_base_hashes_lists_delta_bases_only() {
let base_a = write_blob_via_serialize(b"base alpha content here padding");
let base_b = write_blob_via_serialize(b"base bravo content here padding");
let ha = hash::hash(&base_a);
let hb = hash::hash(&base_b);
let target_a = write_blob_via_serialize(b"base alpha content here PADDED!");
let target_b = write_blob_via_serialize(b"base bravo content here PADDED!");
let stream_a = delta::encode(&base_a, &target_a).unwrap();
let stream_b = delta::encode(&base_b, &target_b).unwrap();
let mut w = PackWriter::new();
w.push_raw(ha, &base_a).unwrap(); w.push_delta(&ha, &stream_a).unwrap();
w.push_delta(&hb, &stream_b).unwrap();
w.push_delta(&ha, &stream_a).unwrap(); let pack = w.finish().unwrap();
let mut bases = delta_base_hashes(&pack).unwrap();
bases.sort_unstable();
let mut expected = vec![ha, hb];
expected.sort_unstable();
assert_eq!(bases, expected);
}
#[test]
fn delta_base_hashes_rejects_bad_magic() {
let mut pack = PackWriter::new().finish().unwrap();
pack[0] = b'X';
assert!(matches!(
delta_base_hashes(&pack),
Err(PackError::InvalidMagic)
));
}
#[test]
fn rejects_raw_payload_that_is_not_canonical_object_without_store_write() {
let payload = b"not a serialized mkit object".to_vec();
let payload_hash = hash::hash(&payload);
let mut body = Vec::new();
body.extend_from_slice(MAGIC);
body.extend_from_slice(&VERSION.to_le_bytes());
body.extend_from_slice(&1u32.to_le_bytes());
body.push(0x00);
let payload_len = u32::try_from(payload.len()).unwrap();
body.extend_from_slice(&payload_len.to_le_bytes());
body.extend_from_slice(&payload);
let pack = finish_pack_body(body);
let (_dir, store) = fresh_store();
let err = PackReader::read(&pack, &store).unwrap_err();
assert!(matches!(err, PackError::InvalidObject(_)), "got {err:?}");
assert!(!store.contains(&payload_hash));
}
#[test]
fn rejects_raw_delta_object_without_store_write() {
let delta = crate::object::Object::Delta(crate::object::Delta {
base_hash: [0xAB; 32],
result_size: 0,
instructions: Vec::new(),
});
let payload = crate::serialize::serialize(&delta).unwrap();
let payload_hash = hash::hash(&payload);
let mut w = PackWriter::new();
w.push_raw(payload_hash, &payload).unwrap();
let pack = w.finish().unwrap();
let (_dir, store) = fresh_store();
let err = PackReader::read(&pack, &store).unwrap_err();
assert!(matches!(err, PackError::NonStorableObject), "got {err:?}");
assert!(!store.contains(&payload_hash));
}
#[test]
fn rejects_delta_resolving_to_non_object_without_partial_store_write() {
let base_obj = write_blob_via_serialize(b"base bytes");
let base_hash = hash::hash(&base_obj);
let invalid_target = b"not a serialized object".to_vec();
let invalid_hash = hash::hash(&invalid_target);
let stream = delta::encode(&base_obj, &invalid_target).unwrap();
let mut w = PackWriter::new();
w.push_raw(base_hash, &base_obj).unwrap();
w.push_delta(&base_hash, &stream).unwrap();
let pack = w.finish().unwrap();
let (_dir, store) = fresh_store();
let err = PackReader::read(&pack, &store).unwrap_err();
assert!(matches!(err, PackError::InvalidObject(_)), "got {err:?}");
assert!(!store.contains(&base_hash));
assert!(!store.contains(&invalid_hash));
}
#[test]
fn rejects_delta_result_over_object_cap_without_partial_store_write() {
let base_obj = write_blob_via_serialize(b"base bytes");
let base_hash = hash::hash(&base_obj);
let mut stream = Vec::new();
stream.push(delta::STREAM_VERSION);
stream.extend_from_slice(&u32::try_from(base_obj.len()).unwrap().to_le_bytes());
stream.extend_from_slice(
&u32::try_from(MAX_RAW_OBJECT_SIZE + 1)
.unwrap()
.to_le_bytes(),
);
let mut w = PackWriter::new();
w.push_raw(base_hash, &base_obj).unwrap();
w.push_delta(&base_hash, &stream).unwrap();
let pack = w.finish().unwrap();
let (_dir, store) = fresh_store();
let err = PackReader::read(&pack, &store).unwrap_err();
assert!(
matches!(
err,
PackError::Store(crate::store::StoreError::ObjectTooLarge)
),
"got {err:?}"
);
assert!(!store.contains(&base_hash));
}
#[test]
fn rejects_trailing_bytes_after_declared_entries_without_store_write() {
let blob = write_blob_via_serialize(b"trailing bytes test");
let blob_hash = hash::hash(&blob);
let mut body = Vec::new();
body.extend_from_slice(MAGIC);
body.extend_from_slice(&VERSION.to_le_bytes());
body.extend_from_slice(&1u32.to_le_bytes());
body.push(0x00);
let blob_len = u32::try_from(blob.len()).unwrap();
body.extend_from_slice(&blob_len.to_le_bytes());
body.extend_from_slice(&blob);
body.extend_from_slice(b"junk");
let pack = finish_pack_body(body);
let (_dir, store) = fresh_store();
let err = PackReader::read(&pack, &store).unwrap_err();
assert!(matches!(err, PackError::TrailingData), "got {err:?}");
assert!(!store.contains(&blob_hash));
}
#[test]
fn rejects_invalid_magic() {
let mut pack = PackWriter::new().finish().unwrap();
pack[0] = b'X';
pack[1] = b'X';
pack[2] = b'X';
pack[3] = b'X';
let (_dir, store) = fresh_store();
let err = PackReader::read(&pack, &store).unwrap_err();
assert!(matches!(err, PackError::InvalidMagic));
}
#[test]
fn rejects_unknown_version() {
let mut pack = PackWriter::new().finish().unwrap();
pack[4] = 99;
let (_dir, store) = fresh_store();
let err = PackReader::read(&pack, &store).unwrap_err();
assert!(matches!(err, PackError::UnsupportedVersion(99)));
}
#[test]
fn rejects_truncated_pack() {
let pack = vec![b'M', b'K']; let (_dir, store) = fresh_store();
let err = PackReader::read(&pack, &store).unwrap_err();
assert!(matches!(err, PackError::PackfileTooShort));
}
#[test]
fn rejects_bit_flipped_trailer() {
let blob = write_blob_via_serialize(b"trailer test");
let h = hash::hash(&blob);
let mut w = PackWriter::new();
w.push_raw(h, &blob).unwrap();
let mut pack = w.finish().unwrap();
let last = pack.len() - 1;
pack[last] ^= 0x01; let (_dir, store) = fresh_store();
let err = PackReader::read(&pack, &store).unwrap_err();
assert!(matches!(err, PackError::PackfileCorrupted));
}
#[test]
fn rejects_reserved_entry_type_0x01() {
let mut buf = Vec::new();
buf.extend_from_slice(MAGIC);
buf.extend_from_slice(&VERSION.to_le_bytes());
buf.extend_from_slice(&1u32.to_le_bytes());
buf.push(0x01); buf.extend_from_slice(&0u32.to_le_bytes()); let trailer = hash::hash(&buf);
buf.extend_from_slice(&trailer);
let (_dir, store) = fresh_store();
let err = PackReader::read(&buf, &store).unwrap_err();
assert!(matches!(err, PackError::InvalidEntryType(0x01)));
}
#[test]
fn rejects_unknown_entry_type() {
let mut buf = Vec::new();
buf.extend_from_slice(MAGIC);
buf.extend_from_slice(&VERSION.to_le_bytes());
buf.extend_from_slice(&1u32.to_le_bytes());
buf.push(0x77); buf.extend_from_slice(&0u32.to_le_bytes());
let trailer = hash::hash(&buf);
buf.extend_from_slice(&trailer);
let (_dir, store) = fresh_store();
let err = PackReader::read(&buf, &store).unwrap_err();
assert!(matches!(err, PackError::InvalidEntryType(0x77)));
}
#[test]
fn delta_base_missing_is_loud() {
let mut fake_base = [0u8; 32];
fake_base[0] = 0xAB;
let mut stream = Vec::new();
stream.push(0x01); stream.extend_from_slice(&0u32.to_le_bytes()); stream.extend_from_slice(&0u32.to_le_bytes()); let mut w = PackWriter::new();
w.push_delta(&fake_base, &stream).unwrap();
let pack = w.finish().unwrap();
let (_dir, store) = fresh_store();
let err = PackReader::read(&pack, &store).unwrap_err();
assert!(matches!(err, PackError::DeltaBaseMissing(_)), "got {err:?}");
}
#[test]
fn entry_payload_past_trailer_rejected() {
let mut buf = Vec::new();
buf.extend_from_slice(MAGIC);
buf.extend_from_slice(&VERSION.to_le_bytes());
buf.extend_from_slice(&1u32.to_le_bytes());
buf.push(0x00);
buf.extend_from_slice(&1_000_000u32.to_le_bytes());
let trailer = hash::hash(&buf);
buf.extend_from_slice(&trailer);
let (_dir, store) = fresh_store();
let err = PackReader::read(&buf, &store).unwrap_err();
assert!(matches!(err, PackError::UnexpectedEof));
}
#[test]
fn entry_count_over_cap_rejected() {
let mut buf = Vec::new();
buf.extend_from_slice(MAGIC);
buf.extend_from_slice(&VERSION.to_le_bytes());
buf.extend_from_slice(&u32::MAX.to_le_bytes());
let trailer = hash::hash(&buf);
buf.extend_from_slice(&trailer);
let (_dir, store) = fresh_store();
let err = PackReader::read(&buf, &store).unwrap_err();
assert!(
matches!(err, PackError::TooManyObjects(_)),
"expected TooManyObjects, got {err:?}"
);
}
#[test]
fn payload_sum_over_cap_is_rejected_before_bounds_or_decode() {
let blob_a = write_blob_via_serialize(&incompressible_bytes(0xA5A5, 64));
let blob_b = write_blob_via_serialize(&incompressible_bytes(0xB6B6, 64));
let mut w = PackWriter::new();
w.push_raw(hash::hash(&blob_a), &blob_a).unwrap();
w.push_raw(hash::hash(&blob_b), &blob_b).unwrap();
let pack = w.finish().unwrap();
let (_dir, store) = fresh_store();
let cap = (blob_a.len() as u64) + 10;
let err = PackReader::read_with_payload_cap(&pack, &store, cap).unwrap_err();
assert!(
matches!(err, PackError::PackfileTooLarge),
"expected PackfileTooLarge, got {err:?}"
);
let report = PackReader::read(&pack, &store).unwrap();
assert_eq!(report.raw_count, 2);
}
#[test]
fn pack_key_is_blake3_of_pack_bytes() {
let blob = write_blob_via_serialize(b"key test");
let h = hash::hash(&blob);
let mut w = PackWriter::new();
w.push_raw(h, &blob).unwrap();
let pack = w.finish().unwrap();
assert_eq!(pack_key(&pack), hash::hash(&pack));
}
#[test]
fn unpack_does_not_recopy_raw_payloads_into_a_second_buffer() {
let mut w = PackWriter::new();
for i in 0u32..64 {
let payload = incompressible_bytes(0x1000_0000 + u64::from(i), 16 * 1024);
let blob = write_blob_via_serialize(&payload);
w.push_raw(hash::hash(&blob), &blob).unwrap();
}
let pack = w.finish().unwrap();
assert!(
pack.len() > 512 * 1024,
"sanity: synthetic pack should be substantial, got {}",
pack.len()
);
assert_eq!(
u32::from_le_bytes(pack[VERSION_OFFSET..VERSION_OFFSET + 4].try_into().unwrap()),
VERSION,
"sanity: incompressible filler must stay an uncompressed v1 pack"
);
let (_dir, store) = fresh_store();
let owned_bytes = AtomicU64::new(0);
let report = PackReader::read_tracking_owned_bytes(&pack, &store, &owned_bytes).unwrap();
assert_eq!(report.raw_count, 64);
assert_eq!(
owned_bytes.load(Ordering::Relaxed),
0,
"an all-raw pack must not allocate a second copy of any entry's payload"
);
}
#[test]
fn unpack_owned_bytes_for_deltas_is_exactly_the_delta_targets_not_the_whole_pack() {
let content_base = incompressible_bytes(0x2BAD_2BAD, 4096);
let base_obj = write_blob_via_serialize(&content_base);
let base_hash = hash::hash(&base_obj);
let mut w = PackWriter::new();
w.push_raw(base_hash, &base_obj).unwrap();
let mut expected_owned = 0u64;
for i in 0u32..10 {
let mut target = content_base.clone();
target[i as usize] ^= 0xFF;
let target_obj = write_blob_via_serialize(&target);
let stream = delta::encode(&base_obj, &target_obj).unwrap();
w.push_delta(&base_hash, &stream).unwrap();
expected_owned += target_obj.len() as u64;
}
let pack = w.finish().unwrap();
let (_dir, store) = fresh_store();
let owned_bytes = AtomicU64::new(0);
let report = PackReader::read_tracking_owned_bytes(&pack, &store, &owned_bytes).unwrap();
assert_eq!(report.raw_count, 1);
assert_eq!(report.delta_count, 10);
assert_eq!(
owned_bytes.load(Ordering::Relaxed),
expected_owned,
"owned bytes must equal exactly the sum of delta target sizes — \
no extra copy of the raw base"
);
}
#[test]
fn pack_writer_finish_does_not_recopy_pushed_payloads() {
let mut w = PackWriter::new();
for i in 0u32..64 {
let payload = incompressible_bytes(0x2000_0000 + u64::from(i), 16 * 1024);
let blob = write_blob_via_serialize(&payload);
w.push_raw(hash::hash(&blob), &blob).unwrap();
}
let bytes_copied = AtomicU64::new(0);
let pack = w.finish_tracking_bytes_copied(&bytes_copied).unwrap();
assert!(pack.len() > 512 * 1024);
assert_eq!(
bytes_copied.load(Ordering::Relaxed),
TRAILER_LEN as u64,
"finish() must only append the trailer, not re-copy every pushed entry"
);
}
#[test]
fn delta_resolves_against_pre_existing_store_object() {
let (_dir, store) = fresh_store();
let mut content_base = vec![0u8; 256];
for (i, b) in content_base.iter_mut().enumerate() {
*b = u8::try_from(i % 251).expect("modulo < 256");
}
let base_obj = write_blob_via_serialize(&content_base);
let base_hash = store.write(&base_obj).unwrap();
let mut content_target = content_base.clone();
content_target[100] = 0xAA;
let target_obj = write_blob_via_serialize(&content_target);
let target_hash = hash::hash(&target_obj);
let stream = delta::encode(&base_obj, &target_obj).unwrap();
let mut w = PackWriter::new();
w.push_delta(&base_hash, &stream).unwrap();
let pack = w.finish().unwrap();
let report = PackReader::read(&pack, &store).unwrap();
assert_eq!(report.delta_count, 1);
assert_eq!(report.raw_count, 0);
assert_eq!(store.read(&target_hash).unwrap(), target_obj);
}
#[test]
fn multiple_deltas_against_shared_external_base_read_store_once() {
const N: usize = 5;
let (_dir, store) = fresh_store();
let mut content_base = vec![0u8; 512];
for (i, b) in content_base.iter_mut().enumerate() {
*b = u8::try_from(i % 251).expect("modulo < 256");
}
let base_obj = write_blob_via_serialize(&content_base);
let base_hash = store.write(&base_obj).unwrap();
let mut w = PackWriter::new();
let mut expected_targets = Vec::new();
for i in 0..N {
let mut content_target = content_base.clone();
content_target[100] = u8::try_from(i).unwrap();
let target_obj = write_blob_via_serialize(&content_target);
let target_hash = hash::hash(&target_obj);
let stream = delta::encode(&base_obj, &target_obj).unwrap();
w.push_delta(&base_hash, &stream).unwrap();
expected_targets.push((target_hash, target_obj));
}
let pack = w.finish().unwrap();
let reads_before = store.read_call_count();
let report = PackReader::read(&pack, &store).unwrap();
let reads_after_for_base = store.read_call_count() - reads_before;
assert_eq!(report.delta_count, u32::try_from(N).unwrap());
assert_eq!(
reads_after_for_base, 1,
"base object must be read from the store exactly once for {N} deltas sharing it, got {reads_after_for_base}"
);
for (target_hash, target_obj) in expected_targets {
assert_eq!(store.read(&target_hash).unwrap(), target_obj);
}
}
fn compressible_bytes(len: usize) -> Vec<u8> {
vec![0x42u8; len]
}
#[test]
#[cfg(feature = "pack-zstd")]
fn compressed_raw_entry_roundtrips() {
let payload = compressible_bytes(4096);
let blob = write_blob_via_serialize(&payload);
let h = hash::hash(&blob);
let mut w = PackWriter::new();
w.push_raw(h, &blob).unwrap();
let pack = w.finish().unwrap();
assert_eq!(
u32::from_le_bytes(pack[VERSION_OFFSET..VERSION_OFFSET + 4].try_into().unwrap()),
VERSION_V2,
"a pack containing a compressed entry must be emitted as version 2"
);
assert_eq!(
pack[HEADER_LEN], 0x03,
"a highly-compressible raw payload must be emitted as 0x03 zstd-raw"
);
let (_dir, store) = fresh_store();
let report = PackReader::read(&pack, &store).unwrap();
assert_eq!(report.raw_count, 1);
assert_eq!(report.delta_count, 0);
assert_eq!(report.stored, vec![h]);
assert_eq!(
store.read(&h).unwrap(),
blob,
"recovered object must be byte-identical to the pre-compression original"
);
}
#[test]
#[cfg(feature = "pack-zstd")]
fn compressed_delta_entry_roundtrips() {
let base_obj =
write_blob_via_serialize(b"delta base filler bytes, not compressible-target-shaped");
let base_hash = hash::hash(&base_obj);
let target_content = compressible_bytes(4096);
let target_obj = write_blob_via_serialize(&target_content);
let target_hash = hash::hash(&target_obj);
let stream = delta::encode(&base_obj, &target_obj).unwrap();
assert!(
stream.len() >= 64,
"sanity: delta stream must clear the writer's compression-candidate floor, got {}",
stream.len()
);
let mut w = PackWriter::new();
w.push_raw(base_hash, &base_obj).unwrap();
w.push_delta(&base_hash, &stream).unwrap();
let pack = w.finish().unwrap();
assert_eq!(
u32::from_le_bytes(pack[VERSION_OFFSET..VERSION_OFFSET + 4].try_into().unwrap()),
VERSION_V2,
"a pack containing a compressed entry must be emitted as version 2"
);
let base_payload_len =
u32::from_le_bytes(pack[HEADER_LEN + 1..HEADER_LEN + 5].try_into().unwrap()) as usize;
let second_entry_type_offset = HEADER_LEN + ENTRY_FRAME_LEN + base_payload_len;
assert_eq!(
pack[second_entry_type_offset], 0x04,
"a highly-compressible delta stream must be emitted as 0x04 zstd-delta"
);
let (_dir, store) = fresh_store();
let report = PackReader::read(&pack, &store).unwrap();
assert_eq!(report.raw_count, 1);
assert_eq!(report.delta_count, 1);
assert_eq!(report.stored, vec![base_hash, target_hash]);
assert_eq!(
store.read(&target_hash).unwrap(),
target_obj,
"recovered delta target must be byte-identical to the pre-compression original"
);
}
#[test]
fn rejects_v2_entry_type_in_v1_pack() {
let mut buf = Vec::new();
buf.extend_from_slice(MAGIC);
buf.extend_from_slice(&VERSION.to_le_bytes()); buf.extend_from_slice(&1u32.to_le_bytes()); buf.push(0x03);
let inner_payload = 0u32.to_le_bytes(); buf.extend_from_slice(&u32::try_from(inner_payload.len()).unwrap().to_le_bytes());
buf.extend_from_slice(&inner_payload);
let pack = finish_pack_body(buf);
let (_dir, store) = fresh_store();
let err = PackReader::read(&pack, &store).unwrap_err();
assert!(
matches!(err, PackError::InvalidEntryType(0x03)),
"got {err:?}"
);
}
#[test]
#[cfg(feature = "pack-zstd")]
fn rejects_decompressed_len_mismatch() {
let payload = compressible_bytes(4096);
let blob = write_blob_via_serialize(&payload);
let h = hash::hash(&blob);
let mut w = PackWriter::new();
w.push_raw(h, &blob).unwrap();
let mut pack = w.finish().unwrap();
assert_eq!(pack[HEADER_LEN], 0x03, "sanity: must be a zstd-raw entry");
let len_prefix_offset = HEADER_LEN + ENTRY_FRAME_LEN;
let claimed_len = u32::from_le_bytes(
pack[len_prefix_offset..len_prefix_offset + 4]
.try_into()
.unwrap(),
);
pack[len_prefix_offset..len_prefix_offset + 4]
.copy_from_slice(&(claimed_len + 1).to_le_bytes());
let split = pack.len() - TRAILER_LEN;
let new_trailer = hash::hash(&pack[..split]);
pack[split..].copy_from_slice(&new_trailer);
let (_dir, store) = fresh_store();
let err = PackReader::read(&pack, &store).unwrap_err();
assert!(
matches!(err, PackError::DecompressedSizeMismatch(_, _)),
"got {err:?}"
);
assert!(!store.contains(&h));
}
#[test]
#[cfg(feature = "pack-zstd")]
fn rejects_decompressed_len_over_object_cap() {
let claimed_len = u32::try_from(MAX_RAW_OBJECT_SIZE + 1).unwrap();
let mut buf = Vec::new();
buf.extend_from_slice(MAGIC);
buf.extend_from_slice(&VERSION_V2.to_le_bytes());
buf.extend_from_slice(&1u32.to_le_bytes());
buf.push(0x03);
let mut inner = Vec::new();
inner.extend_from_slice(&claimed_len.to_le_bytes());
inner.extend_from_slice(&[0u8; 8]);
buf.extend_from_slice(&u32::try_from(inner.len()).unwrap().to_le_bytes());
buf.extend_from_slice(&inner);
let pack = finish_pack_body(buf);
let (_dir, store) = fresh_store();
let err = PackReader::read(&pack, &store).unwrap_err();
assert!(
matches!(err, PackError::DecompressedSizeOverCap(n) if n == claimed_len as usize),
"got {err:?}"
);
}
}