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            objects::fs_atomic::sync_file(&file, &self.path)?;
596        }
597        Ok(())
598    }
599}
600
601pub fn native_pack_excluded_object_types() -> &'static [ObjectType] {
602    &[
603        ObjectType::Redaction,
604        ObjectType::StateVisibility,
605        ObjectType::KeyBinding,
606    ]
607}
608
609pub fn is_native_packable_object_type(obj_type: ObjectType) -> bool {
610    obj_type.packable()
611}
612
613pub fn build_native_pack(
614    store: &impl ObjectStore,
615    objects: &[ObjectInfo],
616) -> Result<NativePackBundle> {
617    let mut builder = PackBuilder::new(sync_pack_compression());
618
619    for info in objects {
620        // Sidecar records (redaction + state-visibility) live outside
621        // `.heddle/objects/` so GC cannot touch them, and must not be
622        // folded into the content-addressed pack. They ship via the
623        // per-object transfer path instead; callers split them out before
624        // packing.
625        if !is_native_packable_object_type(info.obj_type) {
626            continue;
627        }
628        let object = load_object_data(store, &info.id, info.obj_type)?;
629        let pack_id = to_pack_object_id(&object.id);
630        builder.add_id(pack_id, object.obj_type.pack_object_type()?, object.data);
631    }
632
633    let (pack_data, index_data, _) = builder.build()?;
634    Ok(NativePackBundle {
635        pack_data,
636        index_data,
637    })
638}
639
640fn sync_pack_compression() -> CompressionConfig {
641    CompressionConfig {
642        level: 1,
643        min_size: 1024,
644        max_delta_size: 0,
645        ..CompressionConfig::default()
646    }
647}
648
649pub fn install_received_pack(
650    store: &impl ObjectStore,
651    pack_data: &[u8],
652    index_data: &[u8],
653) -> Result<Vec<PackObjectId>> {
654    store
655        .install_pack(pack_data, index_data)
656        .map_err(ProtocolError::from)
657}
658
659pub fn next_pack_chunk(
660    data: &[u8],
661    chunk_size: usize,
662    chunk_index: usize,
663) -> Option<(usize, Vec<u8>, bool)> {
664    let (start, len) = crate::chunk_bounds(data.len(), chunk_size.max(1), chunk_index)?;
665    let is_final = start + len == data.len();
666    Some((start, data[start..start + len].to_vec(), is_final))
667}
668
669pub fn receive_pack_chunk(
670    state: &mut PackChunkState,
671    is_index: bool,
672    resume_offset: u64,
673    chunk_index: u32,
674    is_complete: bool,
675    data: &[u8],
676    is_final_chunk: bool,
677) -> Result<()> {
678    let max_bytes = if is_index {
679        MAX_RECEIVED_PACK_INDEX_SIZE
680    } else {
681        MAX_RECEIVED_PACK_SIZE
682    };
683    receive_pack_chunk_with_limit(
684        state,
685        is_index,
686        resume_offset,
687        chunk_index,
688        is_complete,
689        data,
690        is_final_chunk,
691        max_bytes,
692    )
693}
694
695#[allow(clippy::too_many_arguments)]
696fn receive_pack_chunk_with_limit(
697    state: &mut PackChunkState,
698    is_index: bool,
699    resume_offset: u64,
700    chunk_index: u32,
701    is_complete: bool,
702    data: &[u8],
703    is_final_chunk: bool,
704    max_bytes: u64,
705) -> Result<()> {
706    let (buffer, progress, complete) = if is_index {
707        (
708            &mut state.index_data,
709            &mut state.index_progress,
710            &mut state.index_complete,
711        )
712    } else {
713        (
714            &mut state.pack_data,
715            &mut state.pack_progress,
716            &mut state.pack_complete,
717        )
718    };
719
720    let next_progress = validate_pack_chunk(
721        *progress,
722        is_index,
723        resume_offset,
724        chunk_index,
725        data,
726        max_bytes,
727    )?;
728
729    buffer.extend_from_slice(data);
730    *progress = next_progress;
731    if is_final_chunk || is_complete {
732        *complete = true;
733    }
734    Ok(())
735}
736
737#[allow(clippy::too_many_arguments)]
738fn receive_pack_chunk_to_spool(
739    stream: &mut PackStreamSpool,
740    is_index: bool,
741    resume_offset: u64,
742    chunk_index: u32,
743    is_complete: bool,
744    data: &[u8],
745    is_final_chunk: bool,
746    max_bytes: u64,
747) -> Result<()> {
748    let next_progress = validate_pack_chunk(
749        stream.progress,
750        is_index,
751        resume_offset,
752        chunk_index,
753        data,
754        max_bytes,
755    )?;
756    stream.write_all(data)?;
757    stream.progress = next_progress;
758    if is_final_chunk || is_complete {
759        stream.complete = true;
760    }
761    Ok(())
762}
763
764fn validate_pack_chunk(
765    progress: (u64, u32),
766    is_index: bool,
767    resume_offset: u64,
768    chunk_index: u32,
769    data: &[u8],
770    max_bytes: u64,
771) -> Result<(u64, u32)> {
772    if resume_offset != progress.0 {
773        return Err(ProtocolError::InvalidState(format!(
774            "native pack chunk resume offset mismatch: expected {}, got {}",
775            progress.0, resume_offset
776        )));
777    }
778    if chunk_index != progress.1 {
779        return Err(ProtocolError::InvalidState(format!(
780            "native pack chunk index mismatch: expected {}, got {}",
781            progress.1, chunk_index
782        )));
783    }
784
785    let data_len = u64::try_from(data.len()).map_err(|_| {
786        ProtocolError::InvalidState("native pack chunk length does not fit in u64".to_string())
787    })?;
788    let next_offset = progress.0.checked_add(data_len).ok_or_else(|| {
789        ProtocolError::InvalidState("native pack chunk offset overflow".to_string())
790    })?;
791    if next_offset > max_bytes {
792        let stream_name = if is_index { "index" } else { "body" };
793        return Err(ProtocolError::InvalidState(format!(
794            "native pack {stream_name} exceeds receive size limit: {next_offset} bytes (max {max_bytes})"
795        )));
796    }
797    let next_chunk = progress.1.checked_add(1).ok_or_else(|| {
798        ProtocolError::InvalidState("native pack chunk index overflow".to_string())
799    })?;
800
801    Ok((next_offset, next_chunk))
802}
803
804pub(crate) fn unique_spool_dir(base: &Path) -> Result<PathBuf> {
805    let stamp = SystemTime::now()
806        .duration_since(UNIX_EPOCH)
807        .map_err(|err| {
808            ProtocolError::InvalidState(format!("system clock before UNIX epoch: {err}"))
809        })?
810        .as_nanos();
811    for attempt in 0..100u32 {
812        let dir = base.join(format!("pack-{}-{stamp}-{attempt}", std::process::id()));
813        match fs::create_dir(&dir) {
814            Ok(()) => return Ok(dir),
815            Err(err) if err.kind() == std::io::ErrorKind::AlreadyExists => continue,
816            Err(err) => return Err(ProtocolError::Io(err)),
817        }
818    }
819    Err(ProtocolError::InvalidState(
820        "failed to allocate native pack spool directory".to_string(),
821    ))
822}
823
824fn to_pack_object_id(id: &ObjectId) -> PackObjectId {
825    match id {
826        ObjectId::Hash(hash) => PackObjectId::Hash(*hash),
827        ObjectId::StateId(state_id) => PackObjectId::StateId(*state_id),
828        ObjectId::StateAttachment { id, .. } => PackObjectId::Hash(*id.as_hash()),
829    }
830}
831
832#[cfg(test)]
833mod tests {
834    use objects::{
835        object::{Blob, ContentHash, StateId},
836        store::{
837            CompressionConfig, FsStore, ObjectStore,
838            pack::{ObjectType as PackObjectType, PackBuilder, PackObjectId, PackReader},
839        },
840    };
841    use tempfile::TempDir;
842
843    use super::{
844        GitPackChunkState, GrowingPackChunkReader, MAX_RECEIVED_PACK_SIZE,
845        NativePackStreamingWriter, ObjectData, ObjectId, ObjectInfo, ObjectType, PackChunkSpool,
846        PackChunkState, PackFileChunkReader, build_native_pack, install_received_pack,
847        next_pack_chunk, receive_pack_chunk, receive_pack_chunk_with_limit,
848        reuse_native_pack_encoded_subset_in,
849    };
850
851    fn create_test_store() -> (TempDir, FsStore) {
852        let temp = TempDir::new().unwrap();
853        let store = FsStore::new(temp.path().join(".heddle"));
854        store.init().unwrap();
855        (temp, store)
856    }
857
858    fn hash(byte: u8) -> ContentHash {
859        ContentHash::from_bytes([byte; 32])
860    }
861
862    #[test]
863    fn encoded_snapshot_subset_is_wire_equivalent_without_local_artifacts_or_attachments() {
864        let source = TempDir::new().unwrap();
865        let spool = TempDir::new().unwrap();
866        let source_pack = source.path().join("snapshot.pack");
867        let source_index = source.path().join("snapshot.idx");
868        let blob = (
869            PackObjectId::Hash(hash(1)),
870            PackObjectType::Blob,
871            b"blob body".to_vec(),
872        );
873        let tree = (
874            PackObjectId::Hash(hash(2)),
875            PackObjectType::Tree,
876            b"tree body".to_vec(),
877        );
878        let state_id = StateId::from_bytes([3; 32]);
879        let state = (
880            PackObjectId::StateId(state_id),
881            PackObjectType::State,
882            b"state body".to_vec(),
883        );
884        let attachment_id = PackObjectId::Hash(hash(4));
885        let artifact_id = PackObjectId::Hash(hash(5));
886        let mut builder = PackBuilder::new(CompressionConfig {
887            max_delta_size: 0,
888            ..CompressionConfig::default()
889        });
890        for (id, kind, body) in [
891            blob.clone(),
892            tree.clone(),
893            state.clone(),
894            (
895                attachment_id,
896                PackObjectType::StateAttachment,
897                b"local attachment".to_vec(),
898            ),
899            (
900                artifact_id,
901                PackObjectType::SnapshotCommit,
902                b"local commit artifact".to_vec(),
903            ),
904        ] {
905            builder.add_id(id, kind, body);
906        }
907        let (pack, index, _) = builder.build().unwrap();
908        std::fs::write(&source_pack, pack).unwrap();
909        std::fs::write(&source_index, index).unwrap();
910
911        let wanted = vec![
912            ObjectInfo {
913                id: ObjectId::Hash(hash(1)),
914                obj_type: ObjectType::Blob,
915                size: blob.2.len() as u64,
916                delta_base: None,
917            },
918            ObjectInfo {
919                id: ObjectId::Hash(hash(2)),
920                obj_type: ObjectType::Tree,
921                size: tree.2.len() as u64,
922                delta_base: None,
923            },
924            ObjectInfo {
925                id: ObjectId::StateId(state_id),
926                obj_type: ObjectType::State,
927                size: state.2.len() as u64,
928                delta_base: None,
929            },
930        ];
931        let (bundle, stats) =
932            reuse_native_pack_encoded_subset_in(spool.path(), &source_pack, &wanted)
933                .unwrap()
934                .expect("authoritative non-delta subset must be reusable");
935
936        assert_eq!(stats.object_count, wanted.len());
937        assert!(stats.encoded_bytes_copied > 0);
938        let reused = PackReader::open(&bundle.pack_path, &bundle.index_path).unwrap();
939        let mut reused_ids = reused.list_ids().unwrap();
940        reused_ids.sort();
941        let mut wanted_ids = vec![blob.0, tree.0, state.0];
942        wanted_ids.sort();
943        assert_eq!(reused_ids, wanted_ids);
944        assert!(!reused.has_object(&attachment_id).unwrap());
945        assert!(!reused.has_object(&artifact_id).unwrap());
946
947        for path in [&bundle.pack_path, &bundle.index_path] {
948            let expected_wire_bytes = std::fs::read(path).unwrap();
949            let mut chunk_reader = PackFileChunkReader::open(path, 7).unwrap();
950            let mut wire_bytes = Vec::new();
951            while let Some((offset, chunk_index, data, is_final)) =
952                chunk_reader.next_chunk().unwrap()
953            {
954                assert_eq!(offset as usize, wire_bytes.len());
955                assert_eq!(chunk_index as usize, wire_bytes.len() / 7);
956                wire_bytes.extend_from_slice(&data);
957                assert_eq!(is_final, wire_bytes.len() == expected_wire_bytes.len());
958            }
959            assert_eq!(wire_bytes, expected_wire_bytes);
960        }
961        for (id, _, expected) in [blob, tree, state] {
962            assert_eq!(reused.get_object(&id).unwrap().unwrap().1, expected);
963        }
964    }
965
966    #[test]
967    fn encoded_snapshot_subset_falls_back_for_mismatch_delta_or_attachment_request() {
968        let source = TempDir::new().unwrap();
969        let spool = TempDir::new().unwrap();
970        let source_pack = source.path().join("snapshot.pack");
971        let source_index = source.path().join("snapshot.idx");
972        let first = b"This is the base content. ".repeat(100);
973        let second = b"This is modified content. ".repeat(100);
974        let mut builder = PackBuilder::new(CompressionConfig::default());
975        builder.add(hash(10), PackObjectType::Blob, first.clone());
976        builder.add(hash(11), PackObjectType::Blob, second.clone());
977        let (pack, index, stats) = builder.build().unwrap();
978        assert!(stats.delta_count > 0, "fixture must contain a delta");
979        std::fs::write(&source_pack, pack).unwrap();
980        std::fs::write(&source_index, index).unwrap();
981        let delta_wants = [ObjectInfo {
982            id: ObjectId::Hash(hash(11)),
983            obj_type: ObjectType::Blob,
984            size: second.len() as u64,
985            delta_base: None,
986        }];
987        assert!(
988            reuse_native_pack_encoded_subset_in(spool.path(), &source_pack, &delta_wants)
989                .unwrap()
990                .is_none()
991        );
992
993        let missing_wants = [ObjectInfo {
994            id: ObjectId::Hash(hash(12)),
995            obj_type: ObjectType::Blob,
996            size: 1,
997            delta_base: None,
998        }];
999        assert!(
1000            reuse_native_pack_encoded_subset_in(spool.path(), &source_pack, &missing_wants)
1001                .unwrap()
1002                .is_none()
1003        );
1004
1005        let attachment_wants = [ObjectInfo {
1006            id: ObjectId::StateAttachment {
1007                state: StateId::from_bytes([13; 32]),
1008                id: objects::object::StateAttachmentId::from_hash(hash(14)),
1009                kind: objects::object::StateAttachmentKind::SemanticIndex,
1010            },
1011            obj_type: ObjectType::StateAttachment,
1012            size: 1,
1013            delta_base: None,
1014        }];
1015        assert!(
1016            reuse_native_pack_encoded_subset_in(spool.path(), &source_pack, &attachment_wants)
1017                .unwrap()
1018                .is_none()
1019        );
1020    }
1021
1022    #[test]
1023    fn receive_pack_chunk_rejects_cumulative_size_over_limit_before_buffering() {
1024        let mut state = PackChunkState::default();
1025
1026        receive_pack_chunk_with_limit(&mut state, false, 0, 0, false, b"abcd", false, 8).unwrap();
1027        receive_pack_chunk_with_limit(&mut state, false, 4, 1, false, b"efgh", false, 8).unwrap();
1028
1029        let error = receive_pack_chunk_with_limit(&mut state, false, 8, 2, false, b"i", false, 8)
1030            .unwrap_err();
1031
1032        assert_eq!(state.pack_data, b"abcdefgh");
1033        assert!(
1034            error
1035                .to_string()
1036                .contains("native pack body exceeds receive size limit")
1037        );
1038        assert!(error.to_string().contains("9 bytes (max 8)"));
1039    }
1040
1041    #[test]
1042    fn receive_pack_chunk_checks_production_limit_before_extending_buffer() {
1043        let mut state = PackChunkState {
1044            pack_progress: (MAX_RECEIVED_PACK_SIZE - 1, 0),
1045            ..PackChunkState::default()
1046        };
1047
1048        let error = receive_pack_chunk(
1049            &mut state,
1050            false,
1051            MAX_RECEIVED_PACK_SIZE - 1,
1052            0,
1053            false,
1054            b"xx",
1055            false,
1056        )
1057        .unwrap_err();
1058
1059        assert!(state.pack_data.is_empty());
1060        assert!(
1061            error
1062                .to_string()
1063                .contains("native pack body exceeds receive size limit")
1064        );
1065    }
1066
1067    #[test]
1068    fn receive_pack_chunk_rejects_resume_offset_mismatch_before_buffering() {
1069        let mut state = PackChunkState::default();
1070
1071        let error =
1072            receive_pack_chunk(&mut state, false, 1, 0, false, b"late chunk", false).unwrap_err();
1073
1074        assert!(state.pack_data.is_empty());
1075        assert!(
1076            error
1077                .to_string()
1078                .contains("native pack chunk resume offset mismatch: expected 0, got 1")
1079        );
1080    }
1081
1082    #[test]
1083    fn receive_pack_chunk_rejects_chunk_index_mismatch_before_buffering() {
1084        let mut state = PackChunkState::default();
1085
1086        receive_pack_chunk(&mut state, false, 0, 0, false, b"abc", false).unwrap();
1087        let error = receive_pack_chunk(&mut state, false, 3, 2, false, b"def", false).unwrap_err();
1088
1089        assert_eq!(state.pack_data, b"abc");
1090        assert!(
1091            error
1092                .to_string()
1093                .contains("native pack chunk index mismatch: expected 1, got 2")
1094        );
1095    }
1096
1097    #[test]
1098    fn git_pack_chunk_state_requires_ordered_chunks_and_final_size() {
1099        let mut state = GitPackChunkState::default();
1100
1101        assert!(
1102            state
1103                .receive_chunk("git-pack:test", 0, 0, false, 8, b"abcd")
1104                .unwrap()
1105                .is_none()
1106        );
1107        let error = state
1108            .receive_chunk("git-pack:test", 4, 2, true, 8, b"efgh")
1109            .unwrap_err();
1110
1111        assert!(
1112            error
1113                .to_string()
1114                .contains("Git pack chunk index mismatch: expected 1, got 2")
1115        );
1116        assert!(state.ensure_idle().is_err());
1117
1118        let mut state = GitPackChunkState::default();
1119        state
1120            .receive_chunk("git-pack:test", 0, 0, false, 8, b"abcd")
1121            .unwrap();
1122        let complete = state
1123            .receive_chunk("git-pack:test", 4, 1, true, 8, b"efgh")
1124            .unwrap()
1125            .unwrap();
1126
1127        assert_eq!(complete, b"abcdefgh");
1128        assert!(state.ensure_idle().is_ok());
1129    }
1130
1131    #[test]
1132    fn receive_pack_chunk_accepts_completion_flags_for_pack_and_index() {
1133        let mut state = PackChunkState::default();
1134
1135        receive_pack_chunk(&mut state, false, 0, 0, true, b"pack-body", false).unwrap();
1136        assert!(!state.is_complete());
1137        receive_pack_chunk(&mut state, true, 0, 0, false, b"pack-index", true).unwrap();
1138
1139        assert!(state.is_complete());
1140        assert_eq!(state.pack_data, b"pack-body");
1141        assert_eq!(state.index_data, b"pack-index");
1142    }
1143
1144    #[test]
1145    fn normal_size_native_pack_receives_and_installs() {
1146        let (_source_temp, source_store) = create_test_store();
1147        let (_dest_temp, dest_store) = create_test_store();
1148        let blob = Blob::from("native pack receive regression");
1149        let hash = source_store.put_blob(&blob).unwrap();
1150        let bundle = build_native_pack(
1151            &source_store,
1152            &[ObjectInfo {
1153                id: ObjectId::Hash(hash),
1154                obj_type: ObjectType::Blob,
1155                size: blob.size() as u64,
1156                delta_base: None,
1157            }],
1158        )
1159        .unwrap();
1160
1161        let mut state = PackChunkState::default();
1162        let mut chunk_index = 0usize;
1163        while let Some((start, data, is_final)) = next_pack_chunk(&bundle.pack_data, 7, chunk_index)
1164        {
1165            receive_pack_chunk(
1166                &mut state,
1167                false,
1168                start as u64,
1169                chunk_index as u32,
1170                is_final,
1171                &data,
1172                is_final,
1173            )
1174            .unwrap();
1175            chunk_index += 1;
1176        }
1177
1178        let mut index_chunk = 0usize;
1179        while let Some((start, data, is_final)) =
1180            next_pack_chunk(&bundle.index_data, 5, index_chunk)
1181        {
1182            receive_pack_chunk(
1183                &mut state,
1184                true,
1185                start as u64,
1186                index_chunk as u32,
1187                is_final,
1188                &data,
1189                is_final,
1190            )
1191            .unwrap();
1192            index_chunk += 1;
1193        }
1194
1195        assert!(state.is_complete());
1196        assert_eq!(state.pack_data, bundle.pack_data);
1197        assert_eq!(state.index_data, bundle.index_data);
1198
1199        let installed_ids =
1200            install_received_pack(&dest_store, &state.pack_data, &state.index_data).unwrap();
1201
1202        assert_eq!(installed_ids, vec![PackObjectId::Hash(hash)]);
1203        let installed_blob = dest_store.get_blob(&hash).unwrap().unwrap();
1204        assert_eq!(installed_blob.content(), blob.content());
1205    }
1206
1207    #[test]
1208    fn normal_size_native_pack_spools_and_installs() {
1209        let (_source_temp, source_store) = create_test_store();
1210        let (dest_temp, dest_store) = create_test_store();
1211        let blob = Blob::from("native pack spooled receive regression");
1212        let hash = source_store.put_blob(&blob).unwrap();
1213        let bundle = build_native_pack(
1214            &source_store,
1215            &[ObjectInfo {
1216                id: ObjectId::Hash(hash),
1217                obj_type: ObjectType::Blob,
1218                size: blob.size() as u64,
1219                delta_base: None,
1220            }],
1221        )
1222        .unwrap();
1223
1224        let mut spool = PackChunkSpool::new_in(dest_temp.path()).unwrap();
1225        let mut chunk_index = 0usize;
1226        while let Some((start, data, is_final)) = next_pack_chunk(&bundle.pack_data, 7, chunk_index)
1227        {
1228            spool
1229                .receive_chunk(
1230                    false,
1231                    start as u64,
1232                    chunk_index as u32,
1233                    is_final,
1234                    &data,
1235                    is_final,
1236                )
1237                .unwrap();
1238            chunk_index += 1;
1239        }
1240
1241        let mut index_chunk = 0usize;
1242        while let Some((start, data, is_final)) =
1243            next_pack_chunk(&bundle.index_data, 5, index_chunk)
1244        {
1245            spool
1246                .receive_chunk(
1247                    true,
1248                    start as u64,
1249                    index_chunk as u32,
1250                    is_final,
1251                    &data,
1252                    is_final,
1253                )
1254                .unwrap();
1255            index_chunk += 1;
1256        }
1257
1258        assert!(spool.is_complete());
1259        let installed_ids = spool.install_into(&dest_store).unwrap();
1260
1261        assert_eq!(installed_ids, vec![PackObjectId::Hash(hash)]);
1262        let installed_blob = dest_store.get_blob(&hash).unwrap().unwrap();
1263        assert_eq!(installed_blob.content(), blob.content());
1264    }
1265
1266    #[test]
1267    fn native_pack_streaming_writer_drains_growing_pack_and_installs() {
1268        let (source_temp, source_store) = create_test_store();
1269        let (dest_temp, dest_store) = create_test_store();
1270        let blob = Blob::from("native pack growing stream regression");
1271        let hash = source_store.put_blob(&blob).unwrap();
1272        let large_blob = Blob::from_slice(&vec![b'z'; 4096]);
1273        let large_hash = source_store.put_blob(&large_blob).unwrap();
1274
1275        let mut writer = NativePackStreamingWriter::new_in(source_temp.path(), 2).unwrap();
1276        let mut pack_reader = GrowingPackChunkReader::open(writer.pack_path(), 31).unwrap();
1277        let mut spool = PackChunkSpool::new_in(dest_temp.path()).unwrap();
1278        let mut saw_interleaved_pack_chunk = false;
1279
1280        for (id, obj_type, data) in [
1281            (
1282                ObjectId::Hash(hash),
1283                ObjectType::Blob,
1284                blob.content().to_vec(),
1285            ),
1286            (
1287                ObjectId::Hash(large_hash),
1288                ObjectType::Blob,
1289                large_blob.content().to_vec(),
1290            ),
1291        ] {
1292            writer
1293                .add_object_data(ObjectData {
1294                    id,
1295                    obj_type,
1296                    data,
1297                    is_delta: false,
1298                })
1299                .unwrap();
1300            writer.flush_pack().unwrap();
1301            while let Some((offset, chunk_index, data, is_final)) =
1302                pack_reader.next_available_chunk(false).unwrap()
1303            {
1304                assert!(
1305                    !is_final,
1306                    "pre-final growing pack drain must not mark chunks final"
1307                );
1308                saw_interleaved_pack_chunk = true;
1309                spool
1310                    .receive_chunk(false, offset, chunk_index, false, &data, false)
1311                    .unwrap();
1312            }
1313        }
1314
1315        let bundle = writer.finish().unwrap();
1316        let mut saw_final_pack_chunk = false;
1317        while let Some((offset, chunk_index, data, is_final)) =
1318            pack_reader.next_available_chunk(true).unwrap()
1319        {
1320            saw_final_pack_chunk |= is_final;
1321            spool
1322                .receive_chunk(false, offset, chunk_index, is_final, &data, is_final)
1323                .unwrap();
1324        }
1325
1326        let mut index_reader = PackFileChunkReader::open(&bundle.index_path, 17).unwrap();
1327        while let Some((offset, chunk_index, data, is_final)) = index_reader.next_chunk().unwrap() {
1328            spool
1329                .receive_chunk(true, offset, chunk_index, is_final, &data, is_final)
1330                .unwrap();
1331        }
1332
1333        assert!(
1334            saw_interleaved_pack_chunk,
1335            "expected at least one pack chunk before finalize"
1336        );
1337        assert!(
1338            saw_final_pack_chunk,
1339            "expected final pack chunk after finish"
1340        );
1341        assert!(spool.is_complete());
1342        let mut installed_ids = spool.install_into(&dest_store).unwrap();
1343        let mut expected_ids = vec![PackObjectId::Hash(hash), PackObjectId::Hash(large_hash)];
1344        installed_ids.sort();
1345        expected_ids.sort();
1346
1347        assert_eq!(installed_ids, expected_ids);
1348        let installed_blob = dest_store.get_blob(&hash).unwrap().unwrap();
1349        assert_eq!(installed_blob.content(), blob.content());
1350        let installed_large_blob = dest_store.get_blob(&large_hash).unwrap().unwrap();
1351        assert_eq!(installed_large_blob.content(), large_blob.content());
1352    }
1353}
wire/native_pack.rs

wire/
native_pack.rs
