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

wire/
native_pack.rs
