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, PackReader, 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
60#[derive(Debug, Clone, Copy, PartialEq, Eq)]
61pub struct ReusedNativePackStats {
62    pub object_count: usize,
63    pub encoded_bytes_copied: u64,
64}
65
66/// Build a hosted transport pack by reusing non-delta encoded entries from an
67/// authoritative local pack. `Ok(None)` means the caller must use the normal
68/// object-loading writer.
69pub fn reuse_native_pack_encoded_subset_in(
70    root: &Path,
71    source_pack_path: &Path,
72    objects: &[ObjectInfo],
73) -> Result<Option<(NativePackFileBundle, ReusedNativePackStats)>> {
74    if objects.is_empty()
75        || objects
76            .iter()
77            .any(|object| !object.obj_type.packable_for_push())
78    {
79        return Ok(None);
80    }
81    let source_index_path = source_pack_path.with_extension("idx");
82    if !source_pack_path.is_file() || !source_index_path.is_file() {
83        return Ok(None);
84    }
85    let reader = PackReader::open(source_pack_path, &source_index_path)?;
86    let expected = objects
87        .iter()
88        .map(|object| {
89            Ok((
90                to_pack_object_id(&object.id),
91                object.obj_type.pack_object_type()?,
92                object.size,
93            ))
94        })
95        .collect::<Result<Vec<_>>>()?;
96    let Some(reused) = reader.copy_hosted_encoded_subset(&expected)? else {
97        return Ok(None);
98    };
99
100    let base = root.join("transfer-spool");
101    fs::create_dir_all(&base)?;
102    let dir = unique_spool_dir(&base)?;
103    let pack_path = dir.join("pack");
104    let index_path = dir.join("idx");
105    let write_result = (|| -> Result<(u64, u64)> {
106        fs::write(&pack_path, &reused.pack_data)?;
107        fs::write(&index_path, &reused.index_data)?;
108        Ok((
109            u64::try_from(reused.pack_data.len()).map_err(|_| {
110                ProtocolError::InvalidState("reused pack length exceeds u64".to_string())
111            })?,
112            u64::try_from(reused.index_data.len()).map_err(|_| {
113                ProtocolError::InvalidState("reused pack index length exceeds u64".to_string())
114            })?,
115        ))
116    })();
117    let (pack_len, index_len) = match write_result {
118        Ok(lengths) => lengths,
119        Err(error) => {
120            let _ = fs::remove_dir_all(&dir);
121            return Err(error);
122        }
123    };
124    Ok(Some((
125        NativePackFileBundle {
126            dir,
127            pack_path,
128            index_path,
129            pack_len,
130            index_len,
131        },
132        ReusedNativePackStats {
133            object_count: objects.len(),
134            encoded_bytes_copied: reused.encoded_bytes_copied,
135        },
136    )))
137}
138
139impl Drop for NativePackFileBundle {
140    fn drop(&mut self) {
141        let _ = fs::remove_dir_all(&self.dir);
142    }
143}
144
145#[derive(Debug)]
146pub struct PackFileChunkReader {
147    file: File,
148    total_len: u64,
149    chunk_size: usize,
150    offset: u64,
151    chunk_index: u32,
152}
153
154pub type NativePackFileChunk = (u64, u32, Vec<u8>, bool);
155
156impl PackFileChunkReader {
157    pub fn open(path: &Path, chunk_size: usize) -> Result<Self> {
158        let file = File::open(path)?;
159        let total_len = file.metadata()?.len();
160        Ok(Self {
161            file,
162            total_len,
163            chunk_size: chunk_size.max(1),
164            offset: 0,
165            chunk_index: 0,
166        })
167    }
168
169    pub fn next_chunk(&mut self) -> Result<Option<NativePackFileChunk>> {
170        if self.offset >= self.total_len {
171            return Ok(None);
172        }
173        let remaining = self.total_len - self.offset;
174        let len = remaining.min(self.chunk_size as u64);
175        let len = usize::try_from(len).map_err(|_| {
176            ProtocolError::InvalidState("native pack file chunk length exceeds usize".to_string())
177        })?;
178        let mut data = vec![0u8; len];
179        self.file.read_exact(&mut data)?;
180
181        let offset = self.offset;
182        let chunk_index = self.chunk_index;
183        self.offset = self.offset.checked_add(len as u64).ok_or_else(|| {
184            ProtocolError::InvalidState("native pack file chunk offset overflow".to_string())
185        })?;
186        self.chunk_index = self.chunk_index.checked_add(1).ok_or_else(|| {
187            ProtocolError::InvalidState("native pack file chunk index overflow".to_string())
188        })?;
189        Ok(Some((
190            offset,
191            chunk_index,
192            data,
193            self.offset == self.total_len,
194        )))
195    }
196}
197
198#[derive(Debug)]
199pub struct GrowingPackChunkReader {
200    file: File,
201    chunk_size: usize,
202    offset: u64,
203    chunk_index: u32,
204}
205
206impl GrowingPackChunkReader {
207    pub fn open(path: &Path, chunk_size: usize) -> Result<Self> {
208        Ok(Self {
209            file: File::open(path)?,
210            chunk_size: chunk_size.max(1),
211            offset: 0,
212            chunk_index: 0,
213        })
214    }
215
216    pub fn next_available_chunk(
217        &mut self,
218        final_stream: bool,
219    ) -> Result<Option<NativePackFileChunk>> {
220        let total_len = self.file.metadata()?.len();
221        if self.offset >= total_len {
222            return Ok(None);
223        }
224        let available = total_len - self.offset;
225        if !final_stream && available < self.chunk_size as u64 {
226            return Ok(None);
227        }
228
229        let len = available.min(self.chunk_size as u64);
230        let len = usize::try_from(len).map_err(|_| {
231            ProtocolError::InvalidState(
232                "growing native pack chunk length exceeds usize".to_string(),
233            )
234        })?;
235        let mut data = vec![0u8; len];
236        self.file.read_exact(&mut data)?;
237
238        let offset = self.offset;
239        let chunk_index = self.chunk_index;
240        self.offset = self.offset.checked_add(len as u64).ok_or_else(|| {
241            ProtocolError::InvalidState("growing native pack chunk offset overflow".to_string())
242        })?;
243        self.chunk_index = self.chunk_index.checked_add(1).ok_or_else(|| {
244            ProtocolError::InvalidState("growing native pack chunk index overflow".to_string())
245        })?;
246        Ok(Some((
247            offset,
248            chunk_index,
249            data,
250            final_stream && self.offset == total_len,
251        )))
252    }
253}
254
255pub struct NativePackStreamingWriter {
256    dir: Option<PathBuf>,
257    pack_path: PathBuf,
258    index_path: PathBuf,
259    builder: Option<StreamingPackBuilder<File>>,
260}
261
262impl NativePackStreamingWriter {
263    pub fn new_in(root: &Path, object_count: u64) -> Result<Self> {
264        let base = root.join("transfer-spool");
265        fs::create_dir_all(&base)?;
266        let dir = unique_spool_dir(&base)?;
267        let pack_path = dir.join("pack");
268        let index_path = dir.join("idx");
269        let bucket_dir = dir.join("buckets");
270        let pack_file = OpenOptions::new()
271            .read(true)
272            .write(true)
273            .create_new(true)
274            .open(&pack_path)?;
275        let builder = StreamingPackBuilder::new_with_object_count_ephemeral(
276            pack_file,
277            index_path.clone(),
278            sync_pack_compression(),
279            bucket_dir,
280            object_count,
281        )
282        .map_err(ProtocolError::from)?;
283
284        Ok(Self {
285            dir: Some(dir),
286            pack_path,
287            index_path,
288            builder: Some(builder),
289        })
290    }
291
292    pub fn pack_path(&self) -> &Path {
293        &self.pack_path
294    }
295
296    pub fn index_path(&self) -> &Path {
297        &self.index_path
298    }
299
300    pub fn add_object_data(&mut self, object: ObjectData) -> Result<()> {
301        if !is_native_packable_object_type(object.obj_type) {
302            return Err(ProtocolError::InvalidState(format!(
303                "{:?} sidecar records cannot be packed into the content-addressed object pack",
304                object.obj_type
305            )));
306        }
307        let builder = self.builder.as_mut().ok_or_else(|| {
308            ProtocolError::InvalidState("native pack streaming writer is finalized".to_string())
309        })?;
310        let pack_id = to_pack_object_id(&object.id);
311        builder
312            .add_id(pack_id, object.obj_type.pack_object_type()?, object.data)
313            .map_err(ProtocolError::from)
314    }
315
316    pub fn flush_pack(&mut self) -> Result<()> {
317        let builder = self.builder.as_mut().ok_or_else(|| {
318            ProtocolError::InvalidState("native pack streaming writer is finalized".to_string())
319        })?;
320        builder.flush_pack().map_err(ProtocolError::from)
321    }
322
323    pub fn finish(mut self) -> Result<NativePackFileBundle> {
324        let builder = self.builder.take().ok_or_else(|| {
325            ProtocolError::InvalidState("native pack streaming writer is finalized".to_string())
326        })?;
327        let (mut file, _) = builder.finalize().map_err(ProtocolError::from)?;
328        file.flush()?;
329        drop(file);
330        let pack_len = fs::metadata(&self.pack_path)?.len();
331        let index_len = fs::metadata(&self.index_path)?.len();
332        let dir = self.dir.take().ok_or_else(|| {
333            ProtocolError::InvalidState("native pack streaming writer lost spool dir".to_string())
334        })?;
335        Ok(NativePackFileBundle {
336            dir,
337            pack_path: self.pack_path.clone(),
338            index_path: self.index_path.clone(),
339            pack_len,
340            index_len,
341        })
342    }
343}
344
345impl Drop for NativePackStreamingWriter {
346    fn drop(&mut self) {
347        if let Some(dir) = self.dir.take() {
348            let _ = fs::remove_dir_all(dir);
349        }
350    }
351}
352
353#[derive(Debug, Default, Clone)]
354pub struct PackChunkState {
355    pub pack_data: Vec<u8>,
356    pub index_data: Vec<u8>,
357    pack_progress: (u64, u32),
358    index_progress: (u64, u32),
359    pack_complete: bool,
360    index_complete: bool,
361}
362
363impl PackChunkState {
364    pub fn is_complete(&self) -> bool {
365        self.pack_complete && self.index_complete
366    }
367}
368
369#[derive(Debug, Default, Clone)]
370pub struct GitPackChunkState {
371    transfer_id: Option<String>,
372    pack_size: Option<u64>,
373    next_offset: u64,
374    next_chunk_index: u32,
375    pack_data: Vec<u8>,
376}
377
378impl GitPackChunkState {
379    pub fn is_idle(&self) -> bool {
380        self.transfer_id.is_none()
381            && self.pack_size.is_none()
382            && self.next_offset == 0
383            && self.next_chunk_index == 0
384            && self.pack_data.is_empty()
385    }
386
387    pub fn ensure_idle(&self) -> Result<()> {
388        if self.is_idle() {
389            Ok(())
390        } else {
391            Err(ProtocolError::InvalidState(
392                "Git pack transfer ended before final chunk".to_string(),
393            ))
394        }
395    }
396
397    pub fn receive_chunk(
398        &mut self,
399        transfer_id: &str,
400        offset: u64,
401        chunk_index: u32,
402        is_final_chunk: bool,
403        pack_size: u64,
404        data: &[u8],
405    ) -> Result<Option<Vec<u8>>> {
406        if transfer_id.is_empty() {
407            return Err(ProtocolError::InvalidState(
408                "Git pack transfer_id is required".to_string(),
409            ));
410        }
411        if pack_size > MAX_RECEIVED_GIT_PACK_SIZE {
412            return Err(ProtocolError::InvalidState(format!(
413                "Git pack exceeds maximum transfer size of {MAX_RECEIVED_GIT_PACK_SIZE} bytes"
414            )));
415        }
416        if data.is_empty() {
417            return Err(ProtocolError::InvalidState(
418                "Git pack chunk must not be empty".to_string(),
419            ));
420        }
421        match self.transfer_id.as_ref() {
422            Some(current) if current != transfer_id => {
423                return Err(ProtocolError::InvalidState(format!(
424                    "Git pack transfer id changed from {current:?} to {transfer_id:?}"
425                )));
426            }
427            Some(_) => {}
428            None => {
429                self.transfer_id = Some(transfer_id.to_string());
430                self.pack_size = Some(pack_size);
431            }
432        }
433        if self.pack_size != Some(pack_size) {
434            return Err(ProtocolError::InvalidState(
435                "Git pack size changed during transfer".to_string(),
436            ));
437        }
438        if offset != self.next_offset {
439            return Err(ProtocolError::InvalidState(format!(
440                "Git pack offset mismatch: expected {}, got {}",
441                self.next_offset, offset
442            )));
443        }
444        if chunk_index != self.next_chunk_index {
445            return Err(ProtocolError::InvalidState(format!(
446                "Git pack chunk index mismatch: expected {}, got {}",
447                self.next_chunk_index, chunk_index
448            )));
449        }
450        let chunk_len = u64::try_from(data.len()).map_err(|_| {
451            ProtocolError::InvalidState("Git pack chunk length exceeds u64".to_string())
452        })?;
453        let next_offset = self
454            .next_offset
455            .checked_add(chunk_len)
456            .ok_or_else(|| ProtocolError::InvalidState("Git pack offset overflow".to_string()))?;
457        if next_offset > pack_size {
458            return Err(ProtocolError::InvalidState(
459                "Git pack chunk exceeds declared pack size".to_string(),
460            ));
461        }
462        self.pack_data.extend_from_slice(data);
463        self.next_offset = next_offset;
464        self.next_chunk_index = self.next_chunk_index.checked_add(1).ok_or_else(|| {
465            ProtocolError::InvalidState("Git pack chunk index overflow".to_string())
466        })?;
467        if is_final_chunk {
468            if self.next_offset != pack_size {
469                return Err(ProtocolError::InvalidState(format!(
470                    "Git pack final size mismatch: declared {}, received {}",
471                    pack_size, self.next_offset
472                )));
473            }
474            let pack_data = std::mem::take(&mut self.pack_data);
475            self.transfer_id = None;
476            self.pack_size = None;
477            self.next_offset = 0;
478            self.next_chunk_index = 0;
479            return Ok(Some(pack_data));
480        }
481        if self.next_offset == pack_size {
482            return Err(ProtocolError::InvalidState(
483                "Git pack reached declared size without final chunk marker".to_string(),
484            ));
485        }
486        Ok(None)
487    }
488}
489
490#[derive(Debug)]
491pub struct PackChunkSpool {
492    dir: PathBuf,
493    pack: PackStreamSpool,
494    index: PackStreamSpool,
495}
496
497impl PackChunkSpool {
498    pub fn new_in(root: &Path) -> Result<Self> {
499        let base = root.join("transfer-spool");
500        fs::create_dir_all(&base)?;
501        let dir = unique_spool_dir(&base)?;
502        let pack = PackStreamSpool::new(dir.join("pack"))?;
503        let index = PackStreamSpool::new(dir.join("idx"))?;
504        Ok(Self { dir, pack, index })
505    }
506
507    pub fn is_complete(&self) -> bool {
508        self.pack.complete && self.index.complete
509    }
510
511    #[allow(clippy::too_many_arguments)]
512    pub fn receive_chunk(
513        &mut self,
514        is_index: bool,
515        resume_offset: u64,
516        chunk_index: u32,
517        is_complete: bool,
518        data: &[u8],
519        is_final_chunk: bool,
520    ) -> Result<()> {
521        let max_bytes = if is_index {
522            MAX_RECEIVED_PACK_INDEX_SIZE
523        } else {
524            MAX_RECEIVED_PACK_SIZE
525        };
526        let stream = if is_index {
527            &mut self.index
528        } else {
529            &mut self.pack
530        };
531        receive_pack_chunk_to_spool(
532            stream,
533            is_index,
534            resume_offset,
535            chunk_index,
536            is_complete,
537            data,
538            is_final_chunk,
539            max_bytes,
540        )
541    }
542
543    pub fn install_into(&mut self, store: &impl ObjectStore) -> Result<Vec<PackObjectId>> {
544        if !self.is_complete() {
545            return Err(ProtocolError::InvalidState(
546                "native pack spool is incomplete".to_string(),
547            ));
548        }
549        self.pack.close()?;
550        self.index.close()?;
551        store
552            .install_pack_streaming(&self.pack.path, &self.index.path)
553            .map_err(ProtocolError::from)
554    }
555}
556
557impl Drop for PackChunkSpool {
558    fn drop(&mut self) {
559        let _ = fs::remove_dir_all(&self.dir);
560    }
561}
562
563#[derive(Debug)]
564struct PackStreamSpool {
565    path: PathBuf,
566    file: Option<File>,
567    progress: (u64, u32),
568    complete: bool,
569}
570
571impl PackStreamSpool {
572    fn new(path: PathBuf) -> Result<Self> {
573        let file = File::create(&path)?;
574        Ok(Self {
575            path,
576            file: Some(file),
577            progress: (0, 0),
578            complete: false,
579        })
580    }
581
582    fn write_all(&mut self, data: &[u8]) -> Result<()> {
583        let Some(file) = self.file.as_mut() else {
584            return Err(ProtocolError::InvalidState(
585                "native pack spool stream is already closed".to_string(),
586            ));
587        };
588        file.write_all(data)?;
589        Ok(())
590    }
591
592    fn close(&mut self) -> Result<()> {
593        if let Some(mut file) = self.file.take() {
594            file.flush()?;
595            file.sync_all()?;
596        }
597        Ok(())
598    }
599}
600
601pub fn native_pack_excluded_object_types() -> &'static [ObjectType] {
602    &[ObjectType::Redaction, ObjectType::StateVisibility]
603}
604
605pub fn is_native_packable_object_type(obj_type: ObjectType) -> bool {
606    obj_type.packable()
607}
608
609pub fn build_native_pack(
610    store: &impl ObjectStore,
611    objects: &[ObjectInfo],
612) -> Result<NativePackBundle> {
613    let mut builder = PackBuilder::new(sync_pack_compression());
614
615    for info in objects {
616        // Sidecar records (redaction + state-visibility) live outside
617        // `.heddle/objects/` so GC cannot touch them, and must not be
618        // folded into the content-addressed pack. They ship via the
619        // per-object transfer path instead; callers split them out before
620        // packing.
621        if !is_native_packable_object_type(info.obj_type) {
622            continue;
623        }
624        let object = load_object_data(store, &info.id, info.obj_type)?;
625        let pack_id = to_pack_object_id(&object.id);
626        builder.add_id(pack_id, object.obj_type.pack_object_type()?, object.data);
627    }
628
629    let (pack_data, index_data, _) = builder.build()?;
630    Ok(NativePackBundle {
631        pack_data,
632        index_data,
633    })
634}
635
636fn sync_pack_compression() -> CompressionConfig {
637    CompressionConfig {
638        level: 1,
639        min_size: 1024,
640        max_delta_size: 0,
641        ..CompressionConfig::default()
642    }
643}
644
645pub fn install_received_pack(
646    store: &impl ObjectStore,
647    pack_data: &[u8],
648    index_data: &[u8],
649) -> Result<Vec<PackObjectId>> {
650    store
651        .install_pack(pack_data, index_data)
652        .map_err(ProtocolError::from)
653}
654
655pub fn next_pack_chunk(
656    data: &[u8],
657    chunk_size: usize,
658    chunk_index: usize,
659) -> Option<(usize, Vec<u8>, bool)> {
660    let (start, len) = crate::chunk_bounds(data.len(), chunk_size.max(1), chunk_index)?;
661    let is_final = start + len == data.len();
662    Some((start, data[start..start + len].to_vec(), is_final))
663}
664
665pub fn receive_pack_chunk(
666    state: &mut PackChunkState,
667    is_index: bool,
668    resume_offset: u64,
669    chunk_index: u32,
670    is_complete: bool,
671    data: &[u8],
672    is_final_chunk: bool,
673) -> Result<()> {
674    let max_bytes = if is_index {
675        MAX_RECEIVED_PACK_INDEX_SIZE
676    } else {
677        MAX_RECEIVED_PACK_SIZE
678    };
679    receive_pack_chunk_with_limit(
680        state,
681        is_index,
682        resume_offset,
683        chunk_index,
684        is_complete,
685        data,
686        is_final_chunk,
687        max_bytes,
688    )
689}
690
691#[allow(clippy::too_many_arguments)]
692fn receive_pack_chunk_with_limit(
693    state: &mut PackChunkState,
694    is_index: bool,
695    resume_offset: u64,
696    chunk_index: u32,
697    is_complete: bool,
698    data: &[u8],
699    is_final_chunk: bool,
700    max_bytes: u64,
701) -> Result<()> {
702    let (buffer, progress, complete) = if is_index {
703        (
704            &mut state.index_data,
705            &mut state.index_progress,
706            &mut state.index_complete,
707        )
708    } else {
709        (
710            &mut state.pack_data,
711            &mut state.pack_progress,
712            &mut state.pack_complete,
713        )
714    };
715
716    let next_progress = validate_pack_chunk(
717        *progress,
718        is_index,
719        resume_offset,
720        chunk_index,
721        data,
722        max_bytes,
723    )?;
724
725    buffer.extend_from_slice(data);
726    *progress = next_progress;
727    if is_final_chunk || is_complete {
728        *complete = true;
729    }
730    Ok(())
731}
732
733#[allow(clippy::too_many_arguments)]
734fn receive_pack_chunk_to_spool(
735    stream: &mut PackStreamSpool,
736    is_index: bool,
737    resume_offset: u64,
738    chunk_index: u32,
739    is_complete: bool,
740    data: &[u8],
741    is_final_chunk: bool,
742    max_bytes: u64,
743) -> Result<()> {
744    let next_progress = validate_pack_chunk(
745        stream.progress,
746        is_index,
747        resume_offset,
748        chunk_index,
749        data,
750        max_bytes,
751    )?;
752    stream.write_all(data)?;
753    stream.progress = next_progress;
754    if is_final_chunk || is_complete {
755        stream.complete = true;
756    }
757    Ok(())
758}
759
760fn validate_pack_chunk(
761    progress: (u64, u32),
762    is_index: bool,
763    resume_offset: u64,
764    chunk_index: u32,
765    data: &[u8],
766    max_bytes: u64,
767) -> Result<(u64, u32)> {
768    if resume_offset != progress.0 {
769        return Err(ProtocolError::InvalidState(format!(
770            "native pack chunk resume offset mismatch: expected {}, got {}",
771            progress.0, resume_offset
772        )));
773    }
774    if chunk_index != progress.1 {
775        return Err(ProtocolError::InvalidState(format!(
776            "native pack chunk index mismatch: expected {}, got {}",
777            progress.1, chunk_index
778        )));
779    }
780
781    let data_len = u64::try_from(data.len()).map_err(|_| {
782        ProtocolError::InvalidState("native pack chunk length does not fit in u64".to_string())
783    })?;
784    let next_offset = progress.0.checked_add(data_len).ok_or_else(|| {
785        ProtocolError::InvalidState("native pack chunk offset overflow".to_string())
786    })?;
787    if next_offset > max_bytes {
788        let stream_name = if is_index { "index" } else { "body" };
789        return Err(ProtocolError::InvalidState(format!(
790            "native pack {stream_name} exceeds receive size limit: {next_offset} bytes (max {max_bytes})"
791        )));
792    }
793    let next_chunk = progress.1.checked_add(1).ok_or_else(|| {
794        ProtocolError::InvalidState("native pack chunk index overflow".to_string())
795    })?;
796
797    Ok((next_offset, next_chunk))
798}
799
800pub(crate) fn unique_spool_dir(base: &Path) -> Result<PathBuf> {
801    let stamp = SystemTime::now()
802        .duration_since(UNIX_EPOCH)
803        .map_err(|err| {
804            ProtocolError::InvalidState(format!("system clock before UNIX epoch: {err}"))
805        })?
806        .as_nanos();
807    for attempt in 0..100u32 {
808        let dir = base.join(format!("pack-{}-{stamp}-{attempt}", std::process::id()));
809        match fs::create_dir(&dir) {
810            Ok(()) => return Ok(dir),
811            Err(err) if err.kind() == std::io::ErrorKind::AlreadyExists => continue,
812            Err(err) => return Err(ProtocolError::Io(err)),
813        }
814    }
815    Err(ProtocolError::InvalidState(
816        "failed to allocate native pack spool directory".to_string(),
817    ))
818}
819
820fn to_pack_object_id(id: &ObjectId) -> PackObjectId {
821    match id {
822        ObjectId::Hash(hash) => PackObjectId::Hash(*hash),
823        ObjectId::StateId(state_id) => PackObjectId::StateId(*state_id),
824        ObjectId::StateAttachment { id, .. } => PackObjectId::Hash(*id.as_hash()),
825    }
826}
827
828#[cfg(test)]
829mod tests {
830    use objects::{
831        object::{Blob, ContentHash, StateId},
832        store::{
833            CompressionConfig, FsStore, ObjectStore,
834            pack::{ObjectType as PackObjectType, PackBuilder, PackObjectId, PackReader},
835        },
836    };
837    use tempfile::TempDir;
838
839    use super::{
840        GitPackChunkState, GrowingPackChunkReader, MAX_RECEIVED_PACK_SIZE,
841        NativePackStreamingWriter, ObjectData, ObjectId, ObjectInfo, ObjectType, PackChunkSpool,
842        PackChunkState, PackFileChunkReader, build_native_pack, install_received_pack,
843        next_pack_chunk, receive_pack_chunk, receive_pack_chunk_with_limit,
844        reuse_native_pack_encoded_subset_in,
845    };
846
847    fn create_test_store() -> (TempDir, FsStore) {
848        let temp = TempDir::new().unwrap();
849        let store = FsStore::new(temp.path().join(".heddle"));
850        store.init().unwrap();
851        (temp, store)
852    }
853
854    fn hash(byte: u8) -> ContentHash {
855        ContentHash::from_bytes([byte; 32])
856    }
857
858    #[test]
859    fn encoded_snapshot_subset_is_wire_equivalent_without_local_artifacts_or_attachments() {
860        let source = TempDir::new().unwrap();
861        let spool = TempDir::new().unwrap();
862        let source_pack = source.path().join("snapshot.pack");
863        let source_index = source.path().join("snapshot.idx");
864        let blob = (
865            PackObjectId::Hash(hash(1)),
866            PackObjectType::Blob,
867            b"blob body".to_vec(),
868        );
869        let tree = (
870            PackObjectId::Hash(hash(2)),
871            PackObjectType::Tree,
872            b"tree body".to_vec(),
873        );
874        let state_id = StateId::from_bytes([3; 32]);
875        let state = (
876            PackObjectId::StateId(state_id),
877            PackObjectType::State,
878            b"state body".to_vec(),
879        );
880        let attachment_id = PackObjectId::Hash(hash(4));
881        let artifact_id = PackObjectId::Hash(hash(5));
882        let mut builder = PackBuilder::new(CompressionConfig {
883            max_delta_size: 0,
884            ..CompressionConfig::default()
885        });
886        for (id, kind, body) in [
887            blob.clone(),
888            tree.clone(),
889            state.clone(),
890            (
891                attachment_id,
892                PackObjectType::StateAttachment,
893                b"local attachment".to_vec(),
894            ),
895            (
896                artifact_id,
897                PackObjectType::SnapshotCommit,
898                b"local commit artifact".to_vec(),
899            ),
900        ] {
901            builder.add_id(id, kind, body);
902        }
903        let (pack, index, _) = builder.build().unwrap();
904        std::fs::write(&source_pack, pack).unwrap();
905        std::fs::write(&source_index, index).unwrap();
906
907        let wanted = vec![
908            ObjectInfo {
909                id: ObjectId::Hash(hash(1)),
910                obj_type: ObjectType::Blob,
911                size: blob.2.len() as u64,
912                delta_base: None,
913            },
914            ObjectInfo {
915                id: ObjectId::Hash(hash(2)),
916                obj_type: ObjectType::Tree,
917                size: tree.2.len() as u64,
918                delta_base: None,
919            },
920            ObjectInfo {
921                id: ObjectId::StateId(state_id),
922                obj_type: ObjectType::State,
923                size: state.2.len() as u64,
924                delta_base: None,
925            },
926        ];
927        let (bundle, stats) =
928            reuse_native_pack_encoded_subset_in(spool.path(), &source_pack, &wanted)
929                .unwrap()
930                .expect("authoritative non-delta subset must be reusable");
931
932        assert_eq!(stats.object_count, wanted.len());
933        assert!(stats.encoded_bytes_copied > 0);
934        let reused = PackReader::open(&bundle.pack_path, &bundle.index_path).unwrap();
935        let mut reused_ids = reused.list_ids();
936        reused_ids.sort();
937        let mut wanted_ids = vec![blob.0, tree.0, state.0];
938        wanted_ids.sort();
939        assert_eq!(reused_ids, wanted_ids);
940        assert!(!reused.has_object(&attachment_id));
941        assert!(!reused.has_object(&artifact_id));
942
943        for path in [&bundle.pack_path, &bundle.index_path] {
944            let expected_wire_bytes = std::fs::read(path).unwrap();
945            let mut chunk_reader = PackFileChunkReader::open(path, 7).unwrap();
946            let mut wire_bytes = Vec::new();
947            while let Some((offset, chunk_index, data, is_final)) =
948                chunk_reader.next_chunk().unwrap()
949            {
950                assert_eq!(offset as usize, wire_bytes.len());
951                assert_eq!(chunk_index as usize, wire_bytes.len() / 7);
952                wire_bytes.extend_from_slice(&data);
953                assert_eq!(is_final, wire_bytes.len() == expected_wire_bytes.len());
954            }
955            assert_eq!(wire_bytes, expected_wire_bytes);
956        }
957        for (id, _, expected) in [blob, tree, state] {
958            assert_eq!(reused.get_object(&id).unwrap().unwrap().1, expected);
959        }
960    }
961
962    #[test]
963    fn encoded_snapshot_subset_falls_back_for_mismatch_delta_or_attachment_request() {
964        let source = TempDir::new().unwrap();
965        let spool = TempDir::new().unwrap();
966        let source_pack = source.path().join("snapshot.pack");
967        let source_index = source.path().join("snapshot.idx");
968        let first = b"This is the base content. ".repeat(100);
969        let second = b"This is modified content. ".repeat(100);
970        let mut builder = PackBuilder::new(CompressionConfig::default());
971        builder.add(hash(10), PackObjectType::Blob, first.clone());
972        builder.add(hash(11), PackObjectType::Blob, second.clone());
973        let (pack, index, stats) = builder.build().unwrap();
974        assert!(stats.delta_count > 0, "fixture must contain a delta");
975        std::fs::write(&source_pack, pack).unwrap();
976        std::fs::write(&source_index, index).unwrap();
977        let delta_wants = [ObjectInfo {
978            id: ObjectId::Hash(hash(11)),
979            obj_type: ObjectType::Blob,
980            size: second.len() as u64,
981            delta_base: None,
982        }];
983        assert!(
984            reuse_native_pack_encoded_subset_in(spool.path(), &source_pack, &delta_wants)
985                .unwrap()
986                .is_none()
987        );
988
989        let missing_wants = [ObjectInfo {
990            id: ObjectId::Hash(hash(12)),
991            obj_type: ObjectType::Blob,
992            size: 1,
993            delta_base: None,
994        }];
995        assert!(
996            reuse_native_pack_encoded_subset_in(spool.path(), &source_pack, &missing_wants)
997                .unwrap()
998                .is_none()
999        );
1000
1001        let attachment_wants = [ObjectInfo {
1002            id: ObjectId::StateAttachment {
1003                state: StateId::from_bytes([13; 32]),
1004                id: objects::object::StateAttachmentId::from_hash(hash(14)),
1005                kind: objects::object::StateAttachmentKind::SemanticIndex,
1006            },
1007            obj_type: ObjectType::StateAttachment,
1008            size: 1,
1009            delta_base: None,
1010        }];
1011        assert!(
1012            reuse_native_pack_encoded_subset_in(spool.path(), &source_pack, &attachment_wants)
1013                .unwrap()
1014                .is_none()
1015        );
1016    }
1017
1018    #[test]
1019    fn receive_pack_chunk_rejects_cumulative_size_over_limit_before_buffering() {
1020        let mut state = PackChunkState::default();
1021
1022        receive_pack_chunk_with_limit(&mut state, false, 0, 0, false, b"abcd", false, 8).unwrap();
1023        receive_pack_chunk_with_limit(&mut state, false, 4, 1, false, b"efgh", false, 8).unwrap();
1024
1025        let error = receive_pack_chunk_with_limit(&mut state, false, 8, 2, false, b"i", false, 8)
1026            .unwrap_err();
1027
1028        assert_eq!(state.pack_data, b"abcdefgh");
1029        assert!(
1030            error
1031                .to_string()
1032                .contains("native pack body exceeds receive size limit")
1033        );
1034        assert!(error.to_string().contains("9 bytes (max 8)"));
1035    }
1036
1037    #[test]
1038    fn receive_pack_chunk_checks_production_limit_before_extending_buffer() {
1039        let mut state = PackChunkState {
1040            pack_progress: (MAX_RECEIVED_PACK_SIZE - 1, 0),
1041            ..PackChunkState::default()
1042        };
1043
1044        let error = receive_pack_chunk(
1045            &mut state,
1046            false,
1047            MAX_RECEIVED_PACK_SIZE - 1,
1048            0,
1049            false,
1050            b"xx",
1051            false,
1052        )
1053        .unwrap_err();
1054
1055        assert!(state.pack_data.is_empty());
1056        assert!(
1057            error
1058                .to_string()
1059                .contains("native pack body exceeds receive size limit")
1060        );
1061    }
1062
1063    #[test]
1064    fn receive_pack_chunk_rejects_resume_offset_mismatch_before_buffering() {
1065        let mut state = PackChunkState::default();
1066
1067        let error =
1068            receive_pack_chunk(&mut state, false, 1, 0, false, b"late chunk", false).unwrap_err();
1069
1070        assert!(state.pack_data.is_empty());
1071        assert!(
1072            error
1073                .to_string()
1074                .contains("native pack chunk resume offset mismatch: expected 0, got 1")
1075        );
1076    }
1077
1078    #[test]
1079    fn receive_pack_chunk_rejects_chunk_index_mismatch_before_buffering() {
1080        let mut state = PackChunkState::default();
1081
1082        receive_pack_chunk(&mut state, false, 0, 0, false, b"abc", false).unwrap();
1083        let error = receive_pack_chunk(&mut state, false, 3, 2, false, b"def", false).unwrap_err();
1084
1085        assert_eq!(state.pack_data, b"abc");
1086        assert!(
1087            error
1088                .to_string()
1089                .contains("native pack chunk index mismatch: expected 1, got 2")
1090        );
1091    }
1092
1093    #[test]
1094    fn git_pack_chunk_state_requires_ordered_chunks_and_final_size() {
1095        let mut state = GitPackChunkState::default();
1096
1097        assert!(
1098            state
1099                .receive_chunk("git-pack:test", 0, 0, false, 8, b"abcd")
1100                .unwrap()
1101                .is_none()
1102        );
1103        let error = state
1104            .receive_chunk("git-pack:test", 4, 2, true, 8, b"efgh")
1105            .unwrap_err();
1106
1107        assert!(
1108            error
1109                .to_string()
1110                .contains("Git pack chunk index mismatch: expected 1, got 2")
1111        );
1112        assert!(state.ensure_idle().is_err());
1113
1114        let mut state = GitPackChunkState::default();
1115        state
1116            .receive_chunk("git-pack:test", 0, 0, false, 8, b"abcd")
1117            .unwrap();
1118        let complete = state
1119            .receive_chunk("git-pack:test", 4, 1, true, 8, b"efgh")
1120            .unwrap()
1121            .unwrap();
1122
1123        assert_eq!(complete, b"abcdefgh");
1124        assert!(state.ensure_idle().is_ok());
1125    }
1126
1127    #[test]
1128    fn receive_pack_chunk_accepts_completion_flags_for_pack_and_index() {
1129        let mut state = PackChunkState::default();
1130
1131        receive_pack_chunk(&mut state, false, 0, 0, true, b"pack-body", false).unwrap();
1132        assert!(!state.is_complete());
1133        receive_pack_chunk(&mut state, true, 0, 0, false, b"pack-index", true).unwrap();
1134
1135        assert!(state.is_complete());
1136        assert_eq!(state.pack_data, b"pack-body");
1137        assert_eq!(state.index_data, b"pack-index");
1138    }
1139
1140    #[test]
1141    fn normal_size_native_pack_receives_and_installs() {
1142        let (_source_temp, source_store) = create_test_store();
1143        let (_dest_temp, dest_store) = create_test_store();
1144        let blob = Blob::from("native pack receive regression");
1145        let hash = source_store.put_blob(&blob).unwrap();
1146        let bundle = build_native_pack(
1147            &source_store,
1148            &[ObjectInfo {
1149                id: ObjectId::Hash(hash),
1150                obj_type: ObjectType::Blob,
1151                size: blob.size() as u64,
1152                delta_base: None,
1153            }],
1154        )
1155        .unwrap();
1156
1157        let mut state = PackChunkState::default();
1158        let mut chunk_index = 0usize;
1159        while let Some((start, data, is_final)) = next_pack_chunk(&bundle.pack_data, 7, chunk_index)
1160        {
1161            receive_pack_chunk(
1162                &mut state,
1163                false,
1164                start as u64,
1165                chunk_index as u32,
1166                is_final,
1167                &data,
1168                is_final,
1169            )
1170            .unwrap();
1171            chunk_index += 1;
1172        }
1173
1174        let mut index_chunk = 0usize;
1175        while let Some((start, data, is_final)) =
1176            next_pack_chunk(&bundle.index_data, 5, index_chunk)
1177        {
1178            receive_pack_chunk(
1179                &mut state,
1180                true,
1181                start as u64,
1182                index_chunk as u32,
1183                is_final,
1184                &data,
1185                is_final,
1186            )
1187            .unwrap();
1188            index_chunk += 1;
1189        }
1190
1191        assert!(state.is_complete());
1192        assert_eq!(state.pack_data, bundle.pack_data);
1193        assert_eq!(state.index_data, bundle.index_data);
1194
1195        let installed_ids =
1196            install_received_pack(&dest_store, &state.pack_data, &state.index_data).unwrap();
1197
1198        assert_eq!(installed_ids, vec![PackObjectId::Hash(hash)]);
1199        let installed_blob = dest_store.get_blob(&hash).unwrap().unwrap();
1200        assert_eq!(installed_blob.content(), blob.content());
1201    }
1202
1203    #[test]
1204    fn normal_size_native_pack_spools_and_installs() {
1205        let (_source_temp, source_store) = create_test_store();
1206        let (dest_temp, dest_store) = create_test_store();
1207        let blob = Blob::from("native pack spooled receive regression");
1208        let hash = source_store.put_blob(&blob).unwrap();
1209        let bundle = build_native_pack(
1210            &source_store,
1211            &[ObjectInfo {
1212                id: ObjectId::Hash(hash),
1213                obj_type: ObjectType::Blob,
1214                size: blob.size() as u64,
1215                delta_base: None,
1216            }],
1217        )
1218        .unwrap();
1219
1220        let mut spool = PackChunkSpool::new_in(dest_temp.path()).unwrap();
1221        let mut chunk_index = 0usize;
1222        while let Some((start, data, is_final)) = next_pack_chunk(&bundle.pack_data, 7, chunk_index)
1223        {
1224            spool
1225                .receive_chunk(
1226                    false,
1227                    start as u64,
1228                    chunk_index as u32,
1229                    is_final,
1230                    &data,
1231                    is_final,
1232                )
1233                .unwrap();
1234            chunk_index += 1;
1235        }
1236
1237        let mut index_chunk = 0usize;
1238        while let Some((start, data, is_final)) =
1239            next_pack_chunk(&bundle.index_data, 5, index_chunk)
1240        {
1241            spool
1242                .receive_chunk(
1243                    true,
1244                    start as u64,
1245                    index_chunk as u32,
1246                    is_final,
1247                    &data,
1248                    is_final,
1249                )
1250                .unwrap();
1251            index_chunk += 1;
1252        }
1253
1254        assert!(spool.is_complete());
1255        let installed_ids = spool.install_into(&dest_store).unwrap();
1256
1257        assert_eq!(installed_ids, vec![PackObjectId::Hash(hash)]);
1258        let installed_blob = dest_store.get_blob(&hash).unwrap().unwrap();
1259        assert_eq!(installed_blob.content(), blob.content());
1260    }
1261
1262    #[test]
1263    fn native_pack_streaming_writer_drains_growing_pack_and_installs() {
1264        let (source_temp, source_store) = create_test_store();
1265        let (dest_temp, dest_store) = create_test_store();
1266        let blob = Blob::from("native pack growing stream regression");
1267        let hash = source_store.put_blob(&blob).unwrap();
1268        let large_blob = Blob::from_slice(&vec![b'z'; 4096]);
1269        let large_hash = source_store.put_blob(&large_blob).unwrap();
1270
1271        let mut writer = NativePackStreamingWriter::new_in(source_temp.path(), 2).unwrap();
1272        let mut pack_reader = GrowingPackChunkReader::open(writer.pack_path(), 31).unwrap();
1273        let mut spool = PackChunkSpool::new_in(dest_temp.path()).unwrap();
1274        let mut saw_interleaved_pack_chunk = false;
1275
1276        for (id, obj_type, data) in [
1277            (
1278                ObjectId::Hash(hash),
1279                ObjectType::Blob,
1280                blob.content().to_vec(),
1281            ),
1282            (
1283                ObjectId::Hash(large_hash),
1284                ObjectType::Blob,
1285                large_blob.content().to_vec(),
1286            ),
1287        ] {
1288            writer
1289                .add_object_data(ObjectData {
1290                    id,
1291                    obj_type,
1292                    data,
1293                    is_delta: false,
1294                })
1295                .unwrap();
1296            writer.flush_pack().unwrap();
1297            while let Some((offset, chunk_index, data, is_final)) =
1298                pack_reader.next_available_chunk(false).unwrap()
1299            {
1300                assert!(
1301                    !is_final,
1302                    "pre-final growing pack drain must not mark chunks final"
1303                );
1304                saw_interleaved_pack_chunk = true;
1305                spool
1306                    .receive_chunk(false, offset, chunk_index, false, &data, false)
1307                    .unwrap();
1308            }
1309        }
1310
1311        let bundle = writer.finish().unwrap();
1312        let mut saw_final_pack_chunk = false;
1313        while let Some((offset, chunk_index, data, is_final)) =
1314            pack_reader.next_available_chunk(true).unwrap()
1315        {
1316            saw_final_pack_chunk |= is_final;
1317            spool
1318                .receive_chunk(false, offset, chunk_index, is_final, &data, is_final)
1319                .unwrap();
1320        }
1321
1322        let mut index_reader = PackFileChunkReader::open(&bundle.index_path, 17).unwrap();
1323        while let Some((offset, chunk_index, data, is_final)) = index_reader.next_chunk().unwrap() {
1324            spool
1325                .receive_chunk(true, offset, chunk_index, is_final, &data, is_final)
1326                .unwrap();
1327        }
1328
1329        assert!(
1330            saw_interleaved_pack_chunk,
1331            "expected at least one pack chunk before finalize"
1332        );
1333        assert!(
1334            saw_final_pack_chunk,
1335            "expected final pack chunk after finish"
1336        );
1337        assert!(spool.is_complete());
1338        let mut installed_ids = spool.install_into(&dest_store).unwrap();
1339        let mut expected_ids = vec![PackObjectId::Hash(hash), PackObjectId::Hash(large_hash)];
1340        installed_ids.sort();
1341        expected_ids.sort();
1342
1343        assert_eq!(installed_ids, expected_ids);
1344        let installed_blob = dest_store.get_blob(&hash).unwrap().unwrap();
1345        assert_eq!(installed_blob.content(), blob.content());
1346        let installed_large_blob = dest_store.get_blob(&large_hash).unwrap().unwrap();
1347        assert_eq!(installed_large_blob.content(), large_blob.content());
1348    }
1349}
wire/native_pack.rs

wire/
native_pack.rs
