wire/
native_pack.rs

1// SPDX-License-Identifier: Apache-2.0
2use std::{
3    fs::{self, File, OpenOptions},
4    io::{Read, Write},
5    path::{Path, PathBuf},
6    time::{SystemTime, UNIX_EPOCH},
7};
8
9use objects::store::{
10    CompressionConfig, ObjectStore,
11    pack::{PackBuilder, PackObjectId, StreamingPackBuilder},
12};
13
14use crate::{
15    ObjectData, ObjectId, ObjectInfo, ObjectType, ProtocolError, Result, load_object_data,
16};
17
18/// Maximum hosted native-pack body accepted by the receive primitive.
19///
20/// Native sync packs are produced from bounded state-closure wants and
21/// each decoded pack object is separately capped at 1 GiB in the pack
22/// reader. A 2 GiB compressed pack is materially above normal hosted
23/// sync use while still preventing an untrusted server from growing the
24/// in-memory receive buffer without limit. The receive path can now move
25/// to temp-file spooling plus `install_pack_streaming` — that install API
26/// reports the installed ids the receiver needs, so only the spooling of
27/// the receive buffer itself remains.
28pub const MAX_RECEIVED_PACK_SIZE: u64 = 2 * 1024 * 1024 * 1024;
29
30/// Maximum hosted native-pack index accepted by the receive primitive.
31///
32/// Pack indexes are proportional to object count, not object payload
33/// size. 256 MiB leaves room for millions of entries while bounding the
34/// second in-memory buffer controlled by the remote sender.
35pub const MAX_RECEIVED_PACK_INDEX_SIZE: u64 = 256 * 1024 * 1024;
36
37/// Maximum hosted Git pack accepted by the Git-lane transfer primitive.
38///
39/// Git-overlay sync sends Git-shaped data as raw Git packs. The sender and
40/// receiver still stream those bytes in bounded chunks, but the declared pack
41/// size is untrusted wire input and needs a hard ceiling before buffering or
42/// spooling work begins.
43pub const MAX_RECEIVED_GIT_PACK_SIZE: u64 = 2 * 1024 * 1024 * 1024;
44
45#[derive(Debug, Clone)]
46pub struct NativePackBundle {
47    pub pack_data: Vec<u8>,
48    pub index_data: Vec<u8>,
49}
50
51#[derive(Debug)]
52pub struct NativePackFileBundle {
53    dir: PathBuf,
54    pub pack_path: PathBuf,
55    pub index_path: PathBuf,
56    pub pack_len: u64,
57    pub index_len: u64,
58}
59
60impl Drop for NativePackFileBundle {
61    fn drop(&mut self) {
62        let _ = fs::remove_dir_all(&self.dir);
63    }
64}
65
66#[derive(Debug)]
67pub struct PackFileChunkReader {
68    file: File,
69    total_len: u64,
70    chunk_size: usize,
71    offset: u64,
72    chunk_index: u32,
73}
74
75pub type NativePackFileChunk = (u64, u32, Vec<u8>, bool);
76
77impl PackFileChunkReader {
78    pub fn open(path: &Path, chunk_size: usize) -> Result<Self> {
79        let file = File::open(path)?;
80        let total_len = file.metadata()?.len();
81        Ok(Self {
82            file,
83            total_len,
84            chunk_size: chunk_size.max(1),
85            offset: 0,
86            chunk_index: 0,
87        })
88    }
89
90    pub fn next_chunk(&mut self) -> Result<Option<NativePackFileChunk>> {
91        if self.offset >= self.total_len {
92            return Ok(None);
93        }
94        let remaining = self.total_len - self.offset;
95        let len = remaining.min(self.chunk_size as u64);
96        let len = usize::try_from(len).map_err(|_| {
97            ProtocolError::InvalidState("native pack file chunk length exceeds usize".to_string())
98        })?;
99        let mut data = vec![0u8; len];
100        self.file.read_exact(&mut data)?;
101
102        let offset = self.offset;
103        let chunk_index = self.chunk_index;
104        self.offset = self.offset.checked_add(len as u64).ok_or_else(|| {
105            ProtocolError::InvalidState("native pack file chunk offset overflow".to_string())
106        })?;
107        self.chunk_index = self.chunk_index.checked_add(1).ok_or_else(|| {
108            ProtocolError::InvalidState("native pack file chunk index overflow".to_string())
109        })?;
110        Ok(Some((
111            offset,
112            chunk_index,
113            data,
114            self.offset == self.total_len,
115        )))
116    }
117}
118
119#[derive(Debug)]
120pub struct GrowingPackChunkReader {
121    file: File,
122    chunk_size: usize,
123    offset: u64,
124    chunk_index: u32,
125}
126
127impl GrowingPackChunkReader {
128    pub fn open(path: &Path, chunk_size: usize) -> Result<Self> {
129        Ok(Self {
130            file: File::open(path)?,
131            chunk_size: chunk_size.max(1),
132            offset: 0,
133            chunk_index: 0,
134        })
135    }
136
137    pub fn next_available_chunk(
138        &mut self,
139        final_stream: bool,
140    ) -> Result<Option<NativePackFileChunk>> {
141        let total_len = self.file.metadata()?.len();
142        if self.offset >= total_len {
143            return Ok(None);
144        }
145        let available = total_len - self.offset;
146        if !final_stream && available < self.chunk_size as u64 {
147            return Ok(None);
148        }
149
150        let len = available.min(self.chunk_size as u64);
151        let len = usize::try_from(len).map_err(|_| {
152            ProtocolError::InvalidState(
153                "growing native pack chunk length exceeds usize".to_string(),
154            )
155        })?;
156        let mut data = vec![0u8; len];
157        self.file.read_exact(&mut data)?;
158
159        let offset = self.offset;
160        let chunk_index = self.chunk_index;
161        self.offset = self.offset.checked_add(len as u64).ok_or_else(|| {
162            ProtocolError::InvalidState("growing native pack chunk offset overflow".to_string())
163        })?;
164        self.chunk_index = self.chunk_index.checked_add(1).ok_or_else(|| {
165            ProtocolError::InvalidState("growing native pack chunk index overflow".to_string())
166        })?;
167        Ok(Some((
168            offset,
169            chunk_index,
170            data,
171            final_stream && self.offset == total_len,
172        )))
173    }
174}
175
176pub struct NativePackStreamingWriter {
177    dir: Option<PathBuf>,
178    pack_path: PathBuf,
179    index_path: PathBuf,
180    builder: Option<StreamingPackBuilder<File>>,
181}
182
183impl NativePackStreamingWriter {
184    pub fn new_in(root: &Path, object_count: u64) -> Result<Self> {
185        let base = root.join("transfer-spool");
186        fs::create_dir_all(&base)?;
187        let dir = unique_spool_dir(&base)?;
188        let pack_path = dir.join("pack");
189        let index_path = dir.join("idx");
190        let bucket_dir = dir.join("buckets");
191        let pack_file = OpenOptions::new()
192            .read(true)
193            .write(true)
194            .create_new(true)
195            .open(&pack_path)?;
196        let builder = StreamingPackBuilder::new_with_object_count(
197            pack_file,
198            index_path.clone(),
199            sync_pack_compression(),
200            bucket_dir,
201            object_count,
202        )
203        .map_err(ProtocolError::from)?;
204
205        Ok(Self {
206            dir: Some(dir),
207            pack_path,
208            index_path,
209            builder: Some(builder),
210        })
211    }
212
213    pub fn pack_path(&self) -> &Path {
214        &self.pack_path
215    }
216
217    pub fn index_path(&self) -> &Path {
218        &self.index_path
219    }
220
221    pub fn add_object_data(&mut self, object: ObjectData) -> Result<()> {
222        if !is_native_packable_object_type(object.obj_type) {
223            return Err(ProtocolError::InvalidState(format!(
224                "{:?} sidecar records cannot be packed into the content-addressed object pack",
225                object.obj_type
226            )));
227        }
228        let builder = self.builder.as_mut().ok_or_else(|| {
229            ProtocolError::InvalidState("native pack streaming writer is finalized".to_string())
230        })?;
231        let pack_id = to_pack_object_id(&object.id);
232        builder
233            .add_id(pack_id, object.obj_type.pack_object_type()?, object.data)
234            .map_err(ProtocolError::from)
235    }
236
237    pub fn flush_pack(&mut self) -> Result<()> {
238        let builder = self.builder.as_mut().ok_or_else(|| {
239            ProtocolError::InvalidState("native pack streaming writer is finalized".to_string())
240        })?;
241        builder.flush_pack().map_err(ProtocolError::from)
242    }
243
244    pub fn finish(mut self) -> Result<NativePackFileBundle> {
245        let builder = self.builder.take().ok_or_else(|| {
246            ProtocolError::InvalidState("native pack streaming writer is finalized".to_string())
247        })?;
248        let (file, _) = builder.finalize().map_err(ProtocolError::from)?;
249        file.sync_all()?;
250        drop(file);
251        let pack_len = fs::metadata(&self.pack_path)?.len();
252        let index_len = fs::metadata(&self.index_path)?.len();
253        let dir = self.dir.take().ok_or_else(|| {
254            ProtocolError::InvalidState("native pack streaming writer lost spool dir".to_string())
255        })?;
256        Ok(NativePackFileBundle {
257            dir,
258            pack_path: self.pack_path.clone(),
259            index_path: self.index_path.clone(),
260            pack_len,
261            index_len,
262        })
263    }
264}
265
266impl Drop for NativePackStreamingWriter {
267    fn drop(&mut self) {
268        if let Some(dir) = self.dir.take() {
269            let _ = fs::remove_dir_all(dir);
270        }
271    }
272}
273
274#[derive(Debug, Default, Clone)]
275pub struct PackChunkState {
276    pub pack_data: Vec<u8>,
277    pub index_data: Vec<u8>,
278    pack_progress: (u64, u32),
279    index_progress: (u64, u32),
280    pack_complete: bool,
281    index_complete: bool,
282}
283
284impl PackChunkState {
285    pub fn is_complete(&self) -> bool {
286        self.pack_complete && self.index_complete
287    }
288}
289
290#[derive(Debug, Default, Clone)]
291pub struct GitPackChunkState {
292    transfer_id: Option<String>,
293    pack_size: Option<u64>,
294    next_offset: u64,
295    next_chunk_index: u32,
296    pack_data: Vec<u8>,
297}
298
299impl GitPackChunkState {
300    pub fn is_idle(&self) -> bool {
301        self.transfer_id.is_none()
302            && self.pack_size.is_none()
303            && self.next_offset == 0
304            && self.next_chunk_index == 0
305            && self.pack_data.is_empty()
306    }
307
308    pub fn ensure_idle(&self) -> Result<()> {
309        if self.is_idle() {
310            Ok(())
311        } else {
312            Err(ProtocolError::InvalidState(
313                "Git pack transfer ended before final chunk".to_string(),
314            ))
315        }
316    }
317
318    pub fn receive_chunk(
319        &mut self,
320        transfer_id: &str,
321        offset: u64,
322        chunk_index: u32,
323        is_final_chunk: bool,
324        pack_size: u64,
325        data: &[u8],
326    ) -> Result<Option<Vec<u8>>> {
327        if transfer_id.is_empty() {
328            return Err(ProtocolError::InvalidState(
329                "Git pack transfer_id is required".to_string(),
330            ));
331        }
332        if pack_size > MAX_RECEIVED_GIT_PACK_SIZE {
333            return Err(ProtocolError::InvalidState(format!(
334                "Git pack exceeds maximum transfer size of {MAX_RECEIVED_GIT_PACK_SIZE} bytes"
335            )));
336        }
337        if data.is_empty() {
338            return Err(ProtocolError::InvalidState(
339                "Git pack chunk must not be empty".to_string(),
340            ));
341        }
342        match self.transfer_id.as_ref() {
343            Some(current) if current != transfer_id => {
344                return Err(ProtocolError::InvalidState(format!(
345                    "Git pack transfer id changed from {current:?} to {transfer_id:?}"
346                )));
347            }
348            Some(_) => {}
349            None => {
350                self.transfer_id = Some(transfer_id.to_string());
351                self.pack_size = Some(pack_size);
352            }
353        }
354        if self.pack_size != Some(pack_size) {
355            return Err(ProtocolError::InvalidState(
356                "Git pack size changed during transfer".to_string(),
357            ));
358        }
359        if offset != self.next_offset {
360            return Err(ProtocolError::InvalidState(format!(
361                "Git pack offset mismatch: expected {}, got {}",
362                self.next_offset, offset
363            )));
364        }
365        if chunk_index != self.next_chunk_index {
366            return Err(ProtocolError::InvalidState(format!(
367                "Git pack chunk index mismatch: expected {}, got {}",
368                self.next_chunk_index, chunk_index
369            )));
370        }
371        let chunk_len = u64::try_from(data.len()).map_err(|_| {
372            ProtocolError::InvalidState("Git pack chunk length exceeds u64".to_string())
373        })?;
374        let next_offset = self
375            .next_offset
376            .checked_add(chunk_len)
377            .ok_or_else(|| ProtocolError::InvalidState("Git pack offset overflow".to_string()))?;
378        if next_offset > pack_size {
379            return Err(ProtocolError::InvalidState(
380                "Git pack chunk exceeds declared pack size".to_string(),
381            ));
382        }
383        self.pack_data.extend_from_slice(data);
384        self.next_offset = next_offset;
385        self.next_chunk_index = self.next_chunk_index.checked_add(1).ok_or_else(|| {
386            ProtocolError::InvalidState("Git pack chunk index overflow".to_string())
387        })?;
388        if is_final_chunk {
389            if self.next_offset != pack_size {
390                return Err(ProtocolError::InvalidState(format!(
391                    "Git pack final size mismatch: declared {}, received {}",
392                    pack_size, self.next_offset
393                )));
394            }
395            let pack_data = std::mem::take(&mut self.pack_data);
396            self.transfer_id = None;
397            self.pack_size = None;
398            self.next_offset = 0;
399            self.next_chunk_index = 0;
400            return Ok(Some(pack_data));
401        }
402        if self.next_offset == pack_size {
403            return Err(ProtocolError::InvalidState(
404                "Git pack reached declared size without final chunk marker".to_string(),
405            ));
406        }
407        Ok(None)
408    }
409}
410
411#[derive(Debug)]
412pub struct PackChunkSpool {
413    dir: PathBuf,
414    pack: PackStreamSpool,
415    index: PackStreamSpool,
416}
417
418impl PackChunkSpool {
419    pub fn new_in(root: &Path) -> Result<Self> {
420        let base = root.join("transfer-spool");
421        fs::create_dir_all(&base)?;
422        let dir = unique_spool_dir(&base)?;
423        let pack = PackStreamSpool::new(dir.join("pack"))?;
424        let index = PackStreamSpool::new(dir.join("idx"))?;
425        Ok(Self { dir, pack, index })
426    }
427
428    pub fn is_complete(&self) -> bool {
429        self.pack.complete && self.index.complete
430    }
431
432    #[allow(clippy::too_many_arguments)]
433    pub fn receive_chunk(
434        &mut self,
435        is_index: bool,
436        resume_offset: u64,
437        chunk_index: u32,
438        is_complete: bool,
439        data: &[u8],
440        is_final_chunk: bool,
441    ) -> Result<()> {
442        let max_bytes = if is_index {
443            MAX_RECEIVED_PACK_INDEX_SIZE
444        } else {
445            MAX_RECEIVED_PACK_SIZE
446        };
447        let stream = if is_index {
448            &mut self.index
449        } else {
450            &mut self.pack
451        };
452        receive_pack_chunk_to_spool(
453            stream,
454            is_index,
455            resume_offset,
456            chunk_index,
457            is_complete,
458            data,
459            is_final_chunk,
460            max_bytes,
461        )
462    }
463
464    pub fn install_into(&mut self, store: &impl ObjectStore) -> Result<Vec<PackObjectId>> {
465        if !self.is_complete() {
466            return Err(ProtocolError::InvalidState(
467                "native pack spool is incomplete".to_string(),
468            ));
469        }
470        self.pack.close()?;
471        self.index.close()?;
472        store
473            .install_pack_streaming(&self.pack.path, &self.index.path)
474            .map_err(ProtocolError::from)
475    }
476}
477
478impl Drop for PackChunkSpool {
479    fn drop(&mut self) {
480        let _ = fs::remove_dir_all(&self.dir);
481    }
482}
483
484#[derive(Debug)]
485struct PackStreamSpool {
486    path: PathBuf,
487    file: Option<File>,
488    progress: (u64, u32),
489    complete: bool,
490}
491
492impl PackStreamSpool {
493    fn new(path: PathBuf) -> Result<Self> {
494        let file = File::create(&path)?;
495        Ok(Self {
496            path,
497            file: Some(file),
498            progress: (0, 0),
499            complete: false,
500        })
501    }
502
503    fn write_all(&mut self, data: &[u8]) -> Result<()> {
504        let Some(file) = self.file.as_mut() else {
505            return Err(ProtocolError::InvalidState(
506                "native pack spool stream is already closed".to_string(),
507            ));
508        };
509        file.write_all(data)?;
510        Ok(())
511    }
512
513    fn close(&mut self) -> Result<()> {
514        if let Some(mut file) = self.file.take() {
515            file.flush()?;
516            file.sync_all()?;
517        }
518        Ok(())
519    }
520}
521
522pub fn native_pack_excluded_object_types() -> &'static [ObjectType] {
523    &[ObjectType::Redaction, ObjectType::StateVisibility]
524}
525
526pub fn is_native_packable_object_type(obj_type: ObjectType) -> bool {
527    obj_type.packable()
528}
529
530pub fn build_native_pack(
531    store: &impl ObjectStore,
532    objects: &[ObjectInfo],
533) -> Result<NativePackBundle> {
534    let mut builder = PackBuilder::new(sync_pack_compression());
535
536    for info in objects {
537        // Sidecar records (redaction + state-visibility) live outside
538        // `.heddle/objects/` so GC cannot touch them, and must not be
539        // folded into the content-addressed pack. They ship via the
540        // per-object transfer path instead; callers split them out before
541        // packing.
542        if !is_native_packable_object_type(info.obj_type) {
543            continue;
544        }
545        let object = load_object_data(store, &info.id, info.obj_type)?;
546        let pack_id = to_pack_object_id(&object.id);
547        builder.add_id(pack_id, object.obj_type.pack_object_type()?, object.data);
548    }
549
550    let (pack_data, index_data, _) = builder.build()?;
551    Ok(NativePackBundle {
552        pack_data,
553        index_data,
554    })
555}
556
557fn sync_pack_compression() -> CompressionConfig {
558    CompressionConfig {
559        level: 1,
560        min_size: 1024,
561        max_delta_size: 0,
562        ..CompressionConfig::default()
563    }
564}
565
566pub fn install_received_pack(
567    store: &impl ObjectStore,
568    pack_data: &[u8],
569    index_data: &[u8],
570) -> Result<Vec<PackObjectId>> {
571    store
572        .install_pack(pack_data, index_data)
573        .map_err(ProtocolError::from)
574}
575
576pub fn next_pack_chunk(
577    data: &[u8],
578    chunk_size: usize,
579    chunk_index: usize,
580) -> Option<(usize, Vec<u8>, bool)> {
581    let (start, len) = crate::chunk_bounds(data.len(), chunk_size.max(1), chunk_index)?;
582    let is_final = start + len == data.len();
583    Some((start, data[start..start + len].to_vec(), is_final))
584}
585
586pub fn receive_pack_chunk(
587    state: &mut PackChunkState,
588    is_index: bool,
589    resume_offset: u64,
590    chunk_index: u32,
591    is_complete: bool,
592    data: &[u8],
593    is_final_chunk: bool,
594) -> Result<()> {
595    let max_bytes = if is_index {
596        MAX_RECEIVED_PACK_INDEX_SIZE
597    } else {
598        MAX_RECEIVED_PACK_SIZE
599    };
600    receive_pack_chunk_with_limit(
601        state,
602        is_index,
603        resume_offset,
604        chunk_index,
605        is_complete,
606        data,
607        is_final_chunk,
608        max_bytes,
609    )
610}
611
612#[allow(clippy::too_many_arguments)]
613fn receive_pack_chunk_with_limit(
614    state: &mut PackChunkState,
615    is_index: bool,
616    resume_offset: u64,
617    chunk_index: u32,
618    is_complete: bool,
619    data: &[u8],
620    is_final_chunk: bool,
621    max_bytes: u64,
622) -> Result<()> {
623    let (buffer, progress, complete) = if is_index {
624        (
625            &mut state.index_data,
626            &mut state.index_progress,
627            &mut state.index_complete,
628        )
629    } else {
630        (
631            &mut state.pack_data,
632            &mut state.pack_progress,
633            &mut state.pack_complete,
634        )
635    };
636
637    let next_progress = validate_pack_chunk(
638        *progress,
639        is_index,
640        resume_offset,
641        chunk_index,
642        data,
643        max_bytes,
644    )?;
645
646    buffer.extend_from_slice(data);
647    *progress = next_progress;
648    if is_final_chunk || is_complete {
649        *complete = true;
650    }
651    Ok(())
652}
653
654#[allow(clippy::too_many_arguments)]
655fn receive_pack_chunk_to_spool(
656    stream: &mut PackStreamSpool,
657    is_index: bool,
658    resume_offset: u64,
659    chunk_index: u32,
660    is_complete: bool,
661    data: &[u8],
662    is_final_chunk: bool,
663    max_bytes: u64,
664) -> Result<()> {
665    let next_progress = validate_pack_chunk(
666        stream.progress,
667        is_index,
668        resume_offset,
669        chunk_index,
670        data,
671        max_bytes,
672    )?;
673    stream.write_all(data)?;
674    stream.progress = next_progress;
675    if is_final_chunk || is_complete {
676        stream.complete = true;
677    }
678    Ok(())
679}
680
681fn validate_pack_chunk(
682    progress: (u64, u32),
683    is_index: bool,
684    resume_offset: u64,
685    chunk_index: u32,
686    data: &[u8],
687    max_bytes: u64,
688) -> Result<(u64, u32)> {
689    if resume_offset != progress.0 {
690        return Err(ProtocolError::InvalidState(format!(
691            "native pack chunk resume offset mismatch: expected {}, got {}",
692            progress.0, resume_offset
693        )));
694    }
695    if chunk_index != progress.1 {
696        return Err(ProtocolError::InvalidState(format!(
697            "native pack chunk index mismatch: expected {}, got {}",
698            progress.1, chunk_index
699        )));
700    }
701
702    let data_len = u64::try_from(data.len()).map_err(|_| {
703        ProtocolError::InvalidState("native pack chunk length does not fit in u64".to_string())
704    })?;
705    let next_offset = progress.0.checked_add(data_len).ok_or_else(|| {
706        ProtocolError::InvalidState("native pack chunk offset overflow".to_string())
707    })?;
708    if next_offset > max_bytes {
709        let stream_name = if is_index { "index" } else { "body" };
710        return Err(ProtocolError::InvalidState(format!(
711            "native pack {stream_name} exceeds receive size limit: {next_offset} bytes (max {max_bytes})"
712        )));
713    }
714    let next_chunk = progress.1.checked_add(1).ok_or_else(|| {
715        ProtocolError::InvalidState("native pack chunk index overflow".to_string())
716    })?;
717
718    Ok((next_offset, next_chunk))
719}
720
721fn unique_spool_dir(base: &Path) -> Result<PathBuf> {
722    let stamp = SystemTime::now()
723        .duration_since(UNIX_EPOCH)
724        .map_err(|err| {
725            ProtocolError::InvalidState(format!("system clock before UNIX epoch: {err}"))
726        })?
727        .as_nanos();
728    for attempt in 0..100u32 {
729        let dir = base.join(format!("pack-{}-{stamp}-{attempt}", std::process::id()));
730        match fs::create_dir(&dir) {
731            Ok(()) => return Ok(dir),
732            Err(err) if err.kind() == std::io::ErrorKind::AlreadyExists => continue,
733            Err(err) => return Err(ProtocolError::Io(err)),
734        }
735    }
736    Err(ProtocolError::InvalidState(
737        "failed to allocate native pack spool directory".to_string(),
738    ))
739}
740
741fn to_pack_object_id(id: &ObjectId) -> PackObjectId {
742    match id {
743        ObjectId::Hash(hash) => PackObjectId::Hash(*hash),
744        ObjectId::StateId(state_id) => PackObjectId::StateId(*state_id),
745        ObjectId::StateAttachment { id, .. } => PackObjectId::Hash(*id.as_hash()),
746    }
747}
748
749#[cfg(test)]
750mod tests {
751    use objects::{
752        object::Blob,
753        store::{FsStore, ObjectStore, pack::PackObjectId},
754    };
755    use tempfile::TempDir;
756
757    use super::{
758        GitPackChunkState, GrowingPackChunkReader, MAX_RECEIVED_PACK_SIZE,
759        NativePackStreamingWriter, ObjectData, ObjectId, ObjectInfo, ObjectType, PackChunkSpool,
760        PackChunkState, PackFileChunkReader, build_native_pack, install_received_pack,
761        next_pack_chunk, receive_pack_chunk, receive_pack_chunk_with_limit,
762    };
763
764    fn create_test_store() -> (TempDir, FsStore) {
765        let temp = TempDir::new().unwrap();
766        let store = FsStore::new(temp.path().join(".heddle"));
767        store.init().unwrap();
768        (temp, store)
769    }
770
771    #[test]
772    fn receive_pack_chunk_rejects_cumulative_size_over_limit_before_buffering() {
773        let mut state = PackChunkState::default();
774
775        receive_pack_chunk_with_limit(&mut state, false, 0, 0, false, b"abcd", false, 8).unwrap();
776        receive_pack_chunk_with_limit(&mut state, false, 4, 1, false, b"efgh", false, 8).unwrap();
777
778        let error = receive_pack_chunk_with_limit(&mut state, false, 8, 2, false, b"i", false, 8)
779            .unwrap_err();
780
781        assert_eq!(state.pack_data, b"abcdefgh");
782        assert!(
783            error
784                .to_string()
785                .contains("native pack body exceeds receive size limit")
786        );
787        assert!(error.to_string().contains("9 bytes (max 8)"));
788    }
789
790    #[test]
791    fn receive_pack_chunk_checks_production_limit_before_extending_buffer() {
792        let mut state = PackChunkState {
793            pack_progress: (MAX_RECEIVED_PACK_SIZE - 1, 0),
794            ..PackChunkState::default()
795        };
796
797        let error = receive_pack_chunk(
798            &mut state,
799            false,
800            MAX_RECEIVED_PACK_SIZE - 1,
801            0,
802            false,
803            b"xx",
804            false,
805        )
806        .unwrap_err();
807
808        assert!(state.pack_data.is_empty());
809        assert!(
810            error
811                .to_string()
812                .contains("native pack body exceeds receive size limit")
813        );
814    }
815
816    #[test]
817    fn receive_pack_chunk_rejects_resume_offset_mismatch_before_buffering() {
818        let mut state = PackChunkState::default();
819
820        let error =
821            receive_pack_chunk(&mut state, false, 1, 0, false, b"late chunk", false).unwrap_err();
822
823        assert!(state.pack_data.is_empty());
824        assert!(
825            error
826                .to_string()
827                .contains("native pack chunk resume offset mismatch: expected 0, got 1")
828        );
829    }
830
831    #[test]
832    fn receive_pack_chunk_rejects_chunk_index_mismatch_before_buffering() {
833        let mut state = PackChunkState::default();
834
835        receive_pack_chunk(&mut state, false, 0, 0, false, b"abc", false).unwrap();
836        let error = receive_pack_chunk(&mut state, false, 3, 2, false, b"def", false).unwrap_err();
837
838        assert_eq!(state.pack_data, b"abc");
839        assert!(
840            error
841                .to_string()
842                .contains("native pack chunk index mismatch: expected 1, got 2")
843        );
844    }
845
846    #[test]
847    fn git_pack_chunk_state_requires_ordered_chunks_and_final_size() {
848        let mut state = GitPackChunkState::default();
849
850        assert!(
851            state
852                .receive_chunk("git-pack:test", 0, 0, false, 8, b"abcd")
853                .unwrap()
854                .is_none()
855        );
856        let error = state
857            .receive_chunk("git-pack:test", 4, 2, true, 8, b"efgh")
858            .unwrap_err();
859
860        assert!(
861            error
862                .to_string()
863                .contains("Git pack chunk index mismatch: expected 1, got 2")
864        );
865        assert!(state.ensure_idle().is_err());
866
867        let mut state = GitPackChunkState::default();
868        state
869            .receive_chunk("git-pack:test", 0, 0, false, 8, b"abcd")
870            .unwrap();
871        let complete = state
872            .receive_chunk("git-pack:test", 4, 1, true, 8, b"efgh")
873            .unwrap()
874            .unwrap();
875
876        assert_eq!(complete, b"abcdefgh");
877        assert!(state.ensure_idle().is_ok());
878    }
879
880    #[test]
881    fn receive_pack_chunk_accepts_completion_flags_for_pack_and_index() {
882        let mut state = PackChunkState::default();
883
884        receive_pack_chunk(&mut state, false, 0, 0, true, b"pack-body", false).unwrap();
885        assert!(!state.is_complete());
886        receive_pack_chunk(&mut state, true, 0, 0, false, b"pack-index", true).unwrap();
887
888        assert!(state.is_complete());
889        assert_eq!(state.pack_data, b"pack-body");
890        assert_eq!(state.index_data, b"pack-index");
891    }
892
893    #[test]
894    fn normal_size_native_pack_receives_and_installs() {
895        let (_source_temp, source_store) = create_test_store();
896        let (_dest_temp, dest_store) = create_test_store();
897        let blob = Blob::from("native pack receive regression");
898        let hash = source_store.put_blob(&blob).unwrap();
899        let bundle = build_native_pack(
900            &source_store,
901            &[ObjectInfo {
902                id: ObjectId::Hash(hash),
903                obj_type: ObjectType::Blob,
904                size: blob.size() as u64,
905                delta_base: None,
906            }],
907        )
908        .unwrap();
909
910        let mut state = PackChunkState::default();
911        let mut chunk_index = 0usize;
912        while let Some((start, data, is_final)) = next_pack_chunk(&bundle.pack_data, 7, chunk_index)
913        {
914            receive_pack_chunk(
915                &mut state,
916                false,
917                start as u64,
918                chunk_index as u32,
919                is_final,
920                &data,
921                is_final,
922            )
923            .unwrap();
924            chunk_index += 1;
925        }
926
927        let mut index_chunk = 0usize;
928        while let Some((start, data, is_final)) =
929            next_pack_chunk(&bundle.index_data, 5, index_chunk)
930        {
931            receive_pack_chunk(
932                &mut state,
933                true,
934                start as u64,
935                index_chunk as u32,
936                is_final,
937                &data,
938                is_final,
939            )
940            .unwrap();
941            index_chunk += 1;
942        }
943
944        assert!(state.is_complete());
945        assert_eq!(state.pack_data, bundle.pack_data);
946        assert_eq!(state.index_data, bundle.index_data);
947
948        let installed_ids =
949            install_received_pack(&dest_store, &state.pack_data, &state.index_data).unwrap();
950
951        assert_eq!(installed_ids, vec![PackObjectId::Hash(hash)]);
952        let installed_blob = dest_store.get_blob(&hash).unwrap().unwrap();
953        assert_eq!(installed_blob.content(), blob.content());
954    }
955
956    #[test]
957    fn normal_size_native_pack_spools_and_installs() {
958        let (_source_temp, source_store) = create_test_store();
959        let (dest_temp, dest_store) = create_test_store();
960        let blob = Blob::from("native pack spooled receive regression");
961        let hash = source_store.put_blob(&blob).unwrap();
962        let bundle = build_native_pack(
963            &source_store,
964            &[ObjectInfo {
965                id: ObjectId::Hash(hash),
966                obj_type: ObjectType::Blob,
967                size: blob.size() as u64,
968                delta_base: None,
969            }],
970        )
971        .unwrap();
972
973        let mut spool = PackChunkSpool::new_in(dest_temp.path()).unwrap();
974        let mut chunk_index = 0usize;
975        while let Some((start, data, is_final)) = next_pack_chunk(&bundle.pack_data, 7, chunk_index)
976        {
977            spool
978                .receive_chunk(
979                    false,
980                    start as u64,
981                    chunk_index as u32,
982                    is_final,
983                    &data,
984                    is_final,
985                )
986                .unwrap();
987            chunk_index += 1;
988        }
989
990        let mut index_chunk = 0usize;
991        while let Some((start, data, is_final)) =
992            next_pack_chunk(&bundle.index_data, 5, index_chunk)
993        {
994            spool
995                .receive_chunk(
996                    true,
997                    start as u64,
998                    index_chunk as u32,
999                    is_final,
1000                    &data,
1001                    is_final,
1002                )
1003                .unwrap();
1004            index_chunk += 1;
1005        }
1006
1007        assert!(spool.is_complete());
1008        let installed_ids = spool.install_into(&dest_store).unwrap();
1009
1010        assert_eq!(installed_ids, vec![PackObjectId::Hash(hash)]);
1011        let installed_blob = dest_store.get_blob(&hash).unwrap().unwrap();
1012        assert_eq!(installed_blob.content(), blob.content());
1013    }
1014
1015    #[test]
1016    fn native_pack_streaming_writer_drains_growing_pack_and_installs() {
1017        let (source_temp, source_store) = create_test_store();
1018        let (dest_temp, dest_store) = create_test_store();
1019        let blob = Blob::from("native pack growing stream regression");
1020        let hash = source_store.put_blob(&blob).unwrap();
1021        let large_blob = Blob::from_slice(&vec![b'z'; 4096]);
1022        let large_hash = source_store.put_blob(&large_blob).unwrap();
1023
1024        let mut writer = NativePackStreamingWriter::new_in(source_temp.path(), 2).unwrap();
1025        let mut pack_reader = GrowingPackChunkReader::open(writer.pack_path(), 31).unwrap();
1026        let mut spool = PackChunkSpool::new_in(dest_temp.path()).unwrap();
1027        let mut saw_interleaved_pack_chunk = false;
1028
1029        for (id, obj_type, data) in [
1030            (
1031                ObjectId::Hash(hash),
1032                ObjectType::Blob,
1033                blob.content().to_vec(),
1034            ),
1035            (
1036                ObjectId::Hash(large_hash),
1037                ObjectType::Blob,
1038                large_blob.content().to_vec(),
1039            ),
1040        ] {
1041            writer
1042                .add_object_data(ObjectData {
1043                    id,
1044                    obj_type,
1045                    data,
1046                    is_delta: false,
1047                })
1048                .unwrap();
1049            writer.flush_pack().unwrap();
1050            while let Some((offset, chunk_index, data, is_final)) =
1051                pack_reader.next_available_chunk(false).unwrap()
1052            {
1053                assert!(
1054                    !is_final,
1055                    "pre-final growing pack drain must not mark chunks final"
1056                );
1057                saw_interleaved_pack_chunk = true;
1058                spool
1059                    .receive_chunk(false, offset, chunk_index, false, &data, false)
1060                    .unwrap();
1061            }
1062        }
1063
1064        let bundle = writer.finish().unwrap();
1065        let mut saw_final_pack_chunk = false;
1066        while let Some((offset, chunk_index, data, is_final)) =
1067            pack_reader.next_available_chunk(true).unwrap()
1068        {
1069            saw_final_pack_chunk |= is_final;
1070            spool
1071                .receive_chunk(false, offset, chunk_index, is_final, &data, is_final)
1072                .unwrap();
1073        }
1074
1075        let mut index_reader = PackFileChunkReader::open(&bundle.index_path, 17).unwrap();
1076        while let Some((offset, chunk_index, data, is_final)) = index_reader.next_chunk().unwrap() {
1077            spool
1078                .receive_chunk(true, offset, chunk_index, is_final, &data, is_final)
1079                .unwrap();
1080        }
1081
1082        assert!(
1083            saw_interleaved_pack_chunk,
1084            "expected at least one pack chunk before finalize"
1085        );
1086        assert!(
1087            saw_final_pack_chunk,
1088            "expected final pack chunk after finish"
1089        );
1090        assert!(spool.is_complete());
1091        let mut installed_ids = spool.install_into(&dest_store).unwrap();
1092        let mut expected_ids = vec![PackObjectId::Hash(hash), PackObjectId::Hash(large_hash)];
1093        installed_ids.sort();
1094        expected_ids.sort();
1095
1096        assert_eq!(installed_ids, expected_ids);
1097        let installed_blob = dest_store.get_blob(&hash).unwrap().unwrap();
1098        assert_eq!(installed_blob.content(), blob.content());
1099        let installed_large_blob = dest_store.get_blob(&large_hash).unwrap().unwrap();
1100        assert_eq!(installed_large_blob.content(), large_blob.content());
1101    }
1102}
wire/native_pack.rs

wire/
native_pack.rs
