heddle_pack/store/pack/
pack_reader.rs

1// SPDX-License-Identifier: Apache-2.0
2//! Pack reader for extracting objects from packfiles.
3
4use std::{collections::HashSet, fs::File, io::Read, path::Path};
5
6use bytes::Bytes;
7use heddle_format::delta::{DeltaDecoder, MAX_DELTA_OUTPUT_SIZE};
8
9use super::{
10    ObjectType, PackObjectId, PackObjectRecord, append_container_checksum,
11    decode_tagged_entry_header, decompress_pack_payload, has_zstd_magic, pack_container_spec,
12    pack_index::PackIndex, varint, verify_container, write_container_header,
13};
14use crate::{
15    object::ContentHash,
16    store::{Result, StoreError},
17};
18
19const MAX_PACK_DELTA_OUTPUT_SIZE: usize = MAX_DELTA_OUTPUT_SIZE;
20const MAX_DELTA_CHAIN_DEPTH: usize = 50;
21const MMAP_THRESHOLD_BYTES: u64 = 256 * 1024;
22
23fn read_file_bytes_for_pack(path: &Path) -> Result<Bytes> {
24    let file = File::open(path)?;
25    let len = file.metadata()?.len();
26    if len == 0 {
27        return Ok(Bytes::new());
28    }
29    if len >= MMAP_THRESHOLD_BYTES {
30        let mmap = unsafe { memmap2::MmapOptions::new().map(&file)? };
31        if mmap.len() != checked_file_len_to_usize(len)? {
32            return Err(StoreError::InvalidObject(
33                "pack file size changed during memory mapping".to_string(),
34            ));
35        }
36        return Ok(Bytes::from_owner(mmap));
37    }
38    let mut data = Vec::with_capacity(checked_file_len_to_usize(len)?);
39    let mut reader = file;
40    reader.read_to_end(&mut data)?;
41    Ok(Bytes::from(data))
42}
43
44fn checked_file_len_to_usize(len: u64) -> Result<usize> {
45    usize::try_from(len).map_err(|_| {
46        StoreError::InvalidObject(format!("file length {len} exceeds platform limits"))
47    })
48}
49
50/// Pack reader for extracting objects.
51///
52/// `data` is a refcounted [`Bytes`] view of the pack file. For
53/// uncompressed entries we hand back a zero-copy `Bytes::slice` into
54/// this buffer — no per-blob memcpy, no per-blob allocation. Mmap-
55/// backed `Bytes` (via [`Bytes::from_owner`] on the
56/// `memmap2::Mmap`) survives across reads without copying the
57/// whole pack into the heap.
58enum PackData<'a> {
59    Borrowed(&'a [u8]),
60    Owned(Bytes),
61}
62
63impl<'a> PackData<'a> {
64    fn as_slice(&self) -> &[u8] {
65        match self {
66            Self::Borrowed(data) => data,
67            Self::Owned(data) => data,
68        }
69    }
70
71    fn slice(&self, range: std::ops::Range<usize>) -> Bytes {
72        match self {
73            Self::Borrowed(data) => Bytes::copy_from_slice(&data[range]),
74            Self::Owned(data) => data.slice(range),
75        }
76    }
77}
78
79pub struct PackReader<'a> {
80    data: PackData<'a>,
81    index: PackIndex,
82    content_end: usize,
83}
84
85#[derive(Debug, Clone)]
86pub struct EncodedPackSubset {
87    pub pack_data: Vec<u8>,
88    pub index_data: Vec<u8>,
89    pub encoded_bytes_copied: u64,
90}
91
92impl PackReader<'static> {
93    /// Open a pack file. mmap-backed when the pack is large enough
94    /// to benefit (the same threshold the loose-blob path uses for
95    /// its own mmap decision); read-into-heap otherwise.
96    pub fn open(pack_path: &Path, index_path: &Path) -> Result<Self> {
97        let pack_bytes = read_file_bytes_for_pack(pack_path)?;
98        let index_data = std::fs::read(index_path)?;
99        let (_, _, content_end) = verify_container(&pack_bytes, pack_container_spec())?;
100        let index = PackIndex::from_bytes(&index_data)?;
101        Ok(Self {
102            data: PackData::Owned(pack_bytes),
103            index,
104            content_end,
105        })
106    }
107
108    pub fn from_bytes(pack_data: impl Into<Bytes>, index_data: impl AsRef<[u8]>) -> Result<Self> {
109        let pack_data = pack_data.into();
110        let (_, _, content_end) = verify_container(&pack_data, pack_container_spec())?;
111        let index = PackIndex::from_bytes(index_data.as_ref())?;
112        Ok(Self {
113            data: PackData::Owned(pack_data),
114            index,
115            content_end,
116        })
117    }
118}
119
120impl<'a> PackReader<'a> {
121    pub fn from_slice(pack_data: &'a [u8], index_data: impl AsRef<[u8]>) -> Result<Self> {
122        let (_, _, content_end) = verify_container(pack_data, pack_container_spec())?;
123        let index = PackIndex::from_bytes(index_data.as_ref())?;
124        Ok(Self {
125            data: PackData::Borrowed(pack_data),
126            index,
127            content_end,
128        })
129    }
130
131    /// List all object ids in this pack.
132    pub fn list_ids(&self) -> Vec<PackObjectId> {
133        self.index.ids()
134    }
135
136    pub fn list_hashes(&self) -> Vec<ContentHash> {
137        self.list_ids()
138            .into_iter()
139            .filter_map(|id| match id {
140                PackObjectId::Hash(hash) => Some(hash),
141                PackObjectId::StateId(_) => None,
142            })
143            .collect()
144    }
145
146    pub fn has_object(&self, id: &PackObjectId) -> bool {
147        self.index.find(id).is_some()
148    }
149
150    /// Copy a validated subset of non-delta encoded entries into a standalone
151    /// hosted transport pack without decoding or recompressing their bodies.
152    ///
153    /// `Ok(None)` is a safe fallback signal: an expected object is absent,
154    /// duplicated, has a different type/size, is delta encoded, or names a
155    /// repository-local entry that may not cross the hosted pack boundary.
156    pub fn copy_hosted_encoded_subset(
157        &self,
158        expected: &[(PackObjectId, ObjectType, u64)],
159    ) -> Result<Option<EncodedPackSubset>> {
160        if expected.is_empty() {
161            return Ok(None);
162        }
163        let mut unique = HashSet::with_capacity(expected.len());
164        if expected.iter().any(|(id, obj_type, _)| {
165            !unique.insert(*id)
166                || matches!(
167                    obj_type,
168                    ObjectType::Delta | ObjectType::StateAttachment | ObjectType::SnapshotCommit
169                )
170        }) {
171            return Ok(None);
172        }
173
174        let mut pack_data = Vec::new();
175        write_container_header(&mut pack_data, pack_container_spec(), expected.len() as u64);
176        let mut index = PackIndex::new();
177        let mut encoded_bytes_copied = 0u64;
178        for (expected_id, expected_type, expected_size) in expected {
179            let Some(offset) = self.index.find(expected_id) else {
180                return Ok(None);
181            };
182            let offset = checked_index_offset(offset)?;
183            if offset >= self.content_end {
184                return Err(StoreError::InvalidObject(
185                    "Entry offset out of bounds".to_string(),
186                ));
187            }
188            let header = decode_tagged_entry_header(self.content_from(offset)?)?;
189            let expected_size = usize::try_from(*expected_size).ok();
190            if header.id != *expected_id
191                || header.obj_type != *expected_type
192                || Some(header.uncompressed_size) != expected_size
193                || matches!(
194                    header.obj_type,
195                    ObjectType::Delta | ObjectType::StateAttachment | ObjectType::SnapshotCommit
196                )
197            {
198                return Ok(None);
199            }
200            let encoded_len = header
201                .header_len
202                .checked_add(header.compressed_size)
203                .ok_or_else(|| {
204                    StoreError::InvalidObject("pack entry length overflow".to_string())
205                })?;
206            let encoded_end = offset
207                .checked_add(encoded_len)
208                .ok_or_else(|| StoreError::InvalidObject("pack entry end overflow".to_string()))?;
209            if encoded_end > self.content_end {
210                return Err(StoreError::InvalidObject(
211                    "pack entry extends beyond content boundary".to_string(),
212                ));
213            }
214            let output_offset = u64::try_from(pack_data.len()).map_err(|_| {
215                StoreError::InvalidObject("reused pack offset exceeds u64".to_string())
216            })?;
217            index.add(*expected_id, output_offset);
218            pack_data.extend_from_slice(&self.data.as_slice()[offset..encoded_end]);
219            encoded_bytes_copied = encoded_bytes_copied
220                .checked_add(u64::try_from(encoded_len).map_err(|_| {
221                    StoreError::InvalidObject("encoded pack entry length exceeds u64".to_string())
222                })?)
223                .ok_or_else(|| {
224                    StoreError::InvalidObject("encoded reused byte count overflow".to_string())
225                })?;
226        }
227        index.sort();
228        append_container_checksum(&mut pack_data);
229        Ok(Some(EncodedPackSubset {
230            pack_data,
231            index_data: index.to_bytes(),
232            encoded_bytes_copied,
233        }))
234    }
235
236    /// Get an object from the pack.
237    ///
238    /// Verifies that the tagged id at the indexed offset matches
239    /// `id` before returning. A stale `.idx` file (e.g., overwritten
240    /// in place after a pack rebuild) can otherwise route a request
241    /// for hash `A` to a record physically located at hash `B`'s
242    /// offset — same shape, different content, no error signal.
243    /// This cheap 32-byte id comparison catches that without paying
244    /// a full content-hash recompute on every read; corruption
245    /// strictly *inside* the record body is a separate failure mode
246    /// surfaced via the consumer-side hash verify (see
247    /// `FsStore::loose_blob_path` for the blob equivalent).
248    pub fn get_object(&self, id: &PackObjectId) -> Result<Option<(ObjectType, Vec<u8>)>> {
249        let offset = match self.index.find(id) {
250            Some(offset) => checked_index_offset(offset)?,
251            None => return Ok(None),
252        };
253
254        let record = self.read_record_at_depth(offset, 0)?;
255        verify_record_id_matches(id, &record.id)?;
256        Ok(Some((record.obj_type, record.data)))
257    }
258
259    pub fn get_hashed_object(&self, hash: &ContentHash) -> Result<Option<(ObjectType, Vec<u8>)>> {
260        self.get_object(&PackObjectId::Hash(*hash))
261    }
262
263    /// Zero-copy fast path: when the entry is non-delta and stored
264    /// uncompressed, returns `Bytes::slice` into the pack's
265    /// (mmap-backed) buffer — no allocation, no memcpy. Compressed
266    /// or delta entries fall back to `get_object` and wrap the
267    /// resulting `Vec<u8>` in a `Bytes` (one Arc, no body copy).
268    ///
269    /// Use this from the hot read path. The 10 MB benchmark gap
270    /// between the mount and vanilla FS at the 1 MB+ tier is the
271    /// per-blob memcpy this method eliminates.
272    pub fn get_object_bytes(&self, id: &PackObjectId) -> Result<Option<(ObjectType, Bytes)>> {
273        let Some(offset) = self.index.find(id) else {
274            return Ok(None);
275        };
276        let offset = checked_index_offset(offset)?;
277        if offset >= self.content_end {
278            return Err(StoreError::InvalidObject(
279                "Entry offset out of bounds".to_string(),
280            ));
281        }
282
283        // Verify the tagged id at the indexed offset matches the
284        // requested id — guards against stale-index misrouting (see
285        // `get_object` for the long-form rationale). 32-byte
286        // compare; cheaper than the size+varint decode that follows.
287        let (record_id, id_len) = PackObjectId::decode_tagged(self.content_from(offset)?)?;
288        verify_record_id_matches(id, &record_id)?;
289        let header_start = checked_index_add(offset, id_len, "record header start")?;
290        let (obj_type, uncompressed_size, type_len) =
291            varint::decode_type_and_size(self.content_from(header_start)?).ok_or_else(|| {
292                StoreError::InvalidObject("Truncated type+size varint".to_string())
293            })?;
294        let uncompressed_size = checked_decoded_size("uncompressed_size", uncompressed_size)?;
295        let varint_start = checked_index_add(header_start, type_len, "compressed_size start")?;
296        let (compressed_size, comp_len) = varint::decode_varint(self.content_from(varint_start)?)
297            .ok_or_else(truncated_compressed_size_varint)?;
298        let compressed_size = checked_decoded_size("compressed_size", compressed_size)?;
299
300        // Fast path: non-delta entry stored uncompressed. The most
301        // common shape for snapshot-time packs (the builder skips
302        // the delta search for unrelated blobs).
303        if obj_type != ObjectType::Delta && compressed_size == uncompressed_size {
304            let data_start = checked_index_add(varint_start, comp_len, "entry data start")?;
305            let data_end = checked_data_end(data_start, compressed_size, self.content_end)?;
306            return Ok(Some((obj_type, self.data.slice(data_start..data_end))));
307        }
308
309        // Slow path: defer to the full record reader (it handles
310        // decompression + delta chains) and Bytes-wrap the Vec.
311        // Bytes::from(Vec) is a single Arc allocation, no body copy.
312        let record = self.read_record_at_depth(offset, 0)?;
313        Ok(Some((record.obj_type, Bytes::from(record.data))))
314    }
315
316    pub fn get_hashed_object_bytes(
317        &self,
318        hash: &ContentHash,
319    ) -> Result<Option<(ObjectType, Bytes)>> {
320        self.get_object_bytes(&PackObjectId::Hash(*hash))
321    }
322
323    /// Read just the type+size header for an object without
324    /// decompressing its payload. Returns `Ok(None)` when the object
325    /// isn't in this pack.
326    ///
327    /// For non-delta entries this is one varint decode at the indexed
328    /// offset — much cheaper than `get_object`. Delta entries fall
329    /// back to a full read because their *resolved* size requires
330    /// chasing the base; in practice deltas are rare in the directory
331    /// listing hot path so the fallback is acceptable.
332    pub fn get_hashed_object_size(&self, hash: &ContentHash) -> Result<Option<u64>> {
333        let id = PackObjectId::Hash(*hash);
334        let Some(offset) = self.index.find(&id) else {
335            return Ok(None);
336        };
337        let offset = checked_index_offset(offset)?;
338        if offset >= self.content_end {
339            return Err(StoreError::InvalidObject(
340                "Entry offset out of bounds".to_string(),
341            ));
342        }
343        let (record_id, id_len) = PackObjectId::decode_tagged(self.content_from(offset)?)?;
344        verify_record_id_matches(&id, &record_id)?;
345        let header_start = checked_index_add(offset, id_len, "record header start")?;
346        let (obj_type, uncompressed_size, _type_len) = super::varint::decode_type_and_size(
347            self.content_from(header_start)?,
348        )
349        .ok_or_else(|| StoreError::InvalidObject("Truncated type+size varint".to_string()))?;
350        if obj_type == ObjectType::Delta {
351            // Delta entries record the *resolved* output size in the
352            // type+size varint already (see `read_record_at_depth`'s
353            // size-mismatch check), so we can still return without
354            // decompressing the payload.
355            return Ok(Some(uncompressed_size));
356        }
357        Ok(Some(uncompressed_size))
358    }
359
360    fn read_record_at_depth(&self, offset: usize, depth: usize) -> Result<PackObjectRecord> {
361        if offset >= self.content_end {
362            return Err(StoreError::InvalidObject(
363                "Entry offset out of bounds".to_string(),
364            ));
365        }
366
367        let (id, id_len) = PackObjectId::decode_tagged(self.content_from(offset)?)?;
368        let header_start = checked_index_add(offset, id_len, "record header start")?;
369
370        let (obj_type, uncompressed_size, type_len) =
371            varint::decode_type_and_size(self.content_from(header_start)?).ok_or_else(|| {
372                StoreError::InvalidObject("Truncated type+size varint".to_string())
373            })?;
374        let uncompressed_size = checked_decoded_size("uncompressed_size", uncompressed_size)?;
375
376        let varint_start = checked_index_add(header_start, type_len, "compressed_size start")?;
377        let (compressed_size, comp_len) = varint::decode_varint(self.content_from(varint_start)?)
378            .ok_or_else(truncated_compressed_size_varint)?;
379        let compressed_size = checked_decoded_size("compressed_size", compressed_size)?;
380
381        let mut data_start = checked_index_add(varint_start, comp_len, "entry data start")?;
382
383        // Delta entries carry a tagged base id in pack v2.
384        let base_id = if obj_type == ObjectType::Delta {
385            let (base_id, base_len) = PackObjectId::decode_tagged(self.content_from(data_start)?)?;
386            data_start = checked_index_add(data_start, base_len, "delta data start")?;
387            Some(base_id)
388        } else {
389            None
390        };
391
392        let data_end = checked_data_end(data_start, compressed_size, self.content_end)?;
393
394        let stored_data = &self.data.as_slice()[data_start..data_end];
395
396        // Raw zstd (no wrapper). For non-delta entries, decompress
397        // if sizes differ. For delta entries, the stored data IS the delta
398        // payload (possibly zstd-compressed); check for zstd magic.
399        let decompressed = if obj_type == ObjectType::Delta {
400            if has_zstd_magic(stored_data) {
401                decompress_pack_payload(stored_data, 0)?
402            } else {
403                stored_data.to_vec()
404            }
405        } else if compressed_size != uncompressed_size {
406            decompress_pack_payload(stored_data, uncompressed_size)?
407        } else {
408            stored_data.to_vec()
409        };
410
411        let (resolved_type, final_data) = if obj_type == ObjectType::Delta {
412            self.read_delta_record(base_id, &decompressed, uncompressed_size, depth)?
413        } else {
414            (obj_type, decompressed)
415        };
416
417        if final_data.len() != uncompressed_size {
418            return Err(StoreError::InvalidObject(format!(
419                "Size mismatch: expected {}, got {}",
420                uncompressed_size,
421                final_data.len()
422            )));
423        }
424
425        Ok(PackObjectRecord {
426            id,
427            obj_type: resolved_type,
428            data: final_data,
429            delta_base: None,
430            path_hint: None,
431        })
432    }
433
434    fn read_delta_record(
435        &self,
436        base_id: Option<PackObjectId>,
437        delta: &[u8],
438        uncompressed_size: usize,
439        depth: usize,
440    ) -> Result<(ObjectType, Vec<u8>)> {
441        if depth > MAX_DELTA_CHAIN_DEPTH {
442            return Err(StoreError::InvalidObject(format!(
443                "Delta chain depth {} exceeds max {}",
444                depth, MAX_DELTA_CHAIN_DEPTH
445            )));
446        }
447
448        if uncompressed_size > MAX_PACK_DELTA_OUTPUT_SIZE {
449            return Err(StoreError::InvalidObject(format!(
450                "Delta output size {} exceeds max {}",
451                uncompressed_size, MAX_PACK_DELTA_OUTPUT_SIZE
452            )));
453        }
454
455        let base_hash = Self::require_delta_base_hash(base_id)?;
456        let base_offset = self
457            .index
458            .find(&PackObjectId::Hash(base_hash))
459            .ok_or_else(|| StoreError::NotFound(base_hash.to_string()))?;
460        let base_offset = checked_index_offset(base_offset)?;
461        let base_record = self.read_record_at_depth(base_offset, depth + 1)?;
462        let base_type = base_record.obj_type;
463        let base_data = base_record.data;
464
465        let decoded = DeltaDecoder::decode(&base_data, delta, uncompressed_size)
466            .map_err(|error| StoreError::InvalidObject(format!("Delta decode failed: {error}")))?;
467
468        Ok((base_type, decoded))
469    }
470
471    fn require_delta_base_hash(base_id: Option<PackObjectId>) -> Result<ContentHash> {
472        match base_id {
473            Some(PackObjectId::Hash(hash)) => Ok(hash),
474            Some(PackObjectId::StateId(_)) => Err(StoreError::InvalidObject(
475                "pack delta base must be hash-backed content".into(),
476            )),
477            None => Err(StoreError::InvalidObject(
478                "pack object type is Delta but base hash is missing".into(),
479            )),
480        }
481    }
482
483    fn content_from(&self, offset: usize) -> Result<&[u8]> {
484        if offset > self.content_end {
485            return Err(StoreError::InvalidObject(
486                "Entry header out of bounds".to_string(),
487            ));
488        }
489        Ok(&self.data.as_slice()[offset..self.content_end])
490    }
491}
492
493fn checked_index_offset(offset: u64) -> Result<usize> {
494    usize::try_from(offset)
495        .map_err(|_| StoreError::InvalidObject("Entry offset exceeds platform limits".to_string()))
496}
497
498fn checked_decoded_size(field: &str, size: u64) -> Result<usize> {
499    let size = usize::try_from(size).map_err(|_| {
500        StoreError::InvalidObject(format!("Decoded {field} exceeds platform limits"))
501    })?;
502    if field == "uncompressed_size" && size > super::shared::MAX_PACK_OBJECT_OUTPUT_SIZE {
503        return Err(StoreError::InvalidObject(format!(
504            "Pack object output size {size} exceeds max {}",
505            super::shared::MAX_PACK_OBJECT_OUTPUT_SIZE
506        )));
507    }
508    Ok(size)
509}
510
511fn checked_index_add(start: usize, len: usize, field: &str) -> Result<usize> {
512    start.checked_add(len).ok_or_else(|| {
513        StoreError::InvalidObject(format!("{field} offset overflows platform limits"))
514    })
515}
516
517fn checked_data_end(
518    data_start: usize,
519    compressed_size: usize,
520    content_end: usize,
521) -> Result<usize> {
522    let data_end = data_start.checked_add(compressed_size).ok_or_else(|| {
523        StoreError::InvalidObject("Entry data range overflows platform limits".to_string())
524    })?;
525    if data_end > content_end {
526        return Err(StoreError::InvalidObject(
527            "Entry data out of bounds".to_string(),
528        ));
529    }
530    Ok(data_end)
531}
532
533fn truncated_compressed_size_varint() -> StoreError {
534    StoreError::InvalidObject("Truncated compressed_size varint".to_string())
535}
536
537/// Reject a record whose tagged id at the indexed offset doesn't
538/// match the id the caller asked for. The pack format stores its
539/// records `[tagged_id, type+size, compressed_size, payload]` so the
540/// tagged id is the cheapest available authenticator of "we landed
541/// on the right record"; a stale or hand-edited `.idx` that points
542/// at the *wrong* record produces a mismatch here and we surface it
543/// as a real error instead of silently routing the caller to whatever
544/// bytes happened to be at the bad offset.
545fn verify_record_id_matches(requested: &PackObjectId, found: &PackObjectId) -> Result<()> {
546    if requested == found {
547        return Ok(());
548    }
549    Err(StoreError::InvalidObject(format!(
550        "pack index routed lookup for {requested:?} to record tagged {found:?} \
551         — index is stale or corrupt; the loose-store path will re-promote on \
552         the next read"
553    )))
554}
555
556#[cfg(test)]
557mod tests {
558    use super::{PackObjectId, PackReader, verify_record_id_matches};
559    use crate::{object::ContentHash, store::StoreError};
560
561    #[test]
562    fn test_require_delta_base_hash_rejects_missing_hash() {
563        let error =
564            PackReader::require_delta_base_hash(None).expect_err("missing hash should fail");
565
566        assert!(
567            matches!(error, StoreError::InvalidObject(message) if message == "pack object type is Delta but base hash is missing")
568        );
569    }
570
571    #[test]
572    fn verify_record_id_matches_accepts_identical_ids() {
573        let id = PackObjectId::Hash(ContentHash::from_bytes([7u8; 32]));
574        verify_record_id_matches(&id, &id).expect("matching ids must verify");
575    }
576
577    #[test]
578    fn verify_record_id_matches_rejects_mismatched_ids() {
579        let asked = PackObjectId::Hash(ContentHash::from_bytes([7u8; 32]));
580        let found = PackObjectId::Hash(ContentHash::from_bytes([8u8; 32]));
581        let error = verify_record_id_matches(&asked, &found)
582            .expect_err("mismatched record id must error rather than silently route");
583        assert!(
584            matches!(&error, StoreError::InvalidObject(message) if message.contains("stale or corrupt")),
585            "stale-index mismatch must surface as InvalidObject with the diagnostic phrase, got: {error:?}",
586        );
587    }
588}
heddle_pack/store/pack/pack_reader.rs

heddle_pack/store/pack/
pack_reader.rs
