rust_ipfs/dag/
car.rs

1use super::IpldDag;
2use crate::{Block, Error};
3use anyhow::anyhow;
4use bytes::{Bytes, BytesMut};
5use futures::future::BoxFuture;
6use futures::io::{AsyncRead, AsyncReadExt, BufReader};
7use futures::stream::BoxStream;
8use futures::{FutureExt, Stream, StreamExt, TryStreamExt};
9use ipld_core::cid::Cid;
10use ipld_core::codec::Codec;
11use ipld_core::ipld::Ipld;
12use std::collections::{BTreeMap, HashSet};
13use std::future::IntoFuture;
14#[cfg(not(target_arch = "wasm32"))]
15use std::path::Path;
16use std::pin::Pin;
17use std::task::{Context, Poll};
18use std::time::Duration;
19#[cfg(not(target_arch = "wasm32"))]
20use tokio::io::AsyncWriteExt;
21
22/// The fixed 11-byte CARv2 pragma: `varint(10)` followed by the dag-cbor `{"version": 2}`.
23const CAR_V2_PRAGMA: [u8; 11] = [
24    0x0a, 0xa1, 0x67, 0x76, 0x65, 0x72, 0x73, 0x69, 0x6f, 0x6e, 0x02,
25];
26const V2_HEADER_LEN: usize = 40;
27const IMPORT_BATCH: usize = 256;
28const FETCH_TIMEOUT: Duration = Duration::from_secs(60);
29/// Caps on the length a single CAR varint may declare before we allocate, so a crafted archive
30/// cannot make us reserve gigabytes.
31const MAX_HEADER_LEN: u64 = 32 << 20;
32const MAX_SECTION_LEN: u64 = 32 << 20;
33
34impl IpldDag {
35    /// Exports the DAG rooted at `root` as a CARv1 archive.
36    pub fn export(&self, root: Cid) -> CarExport {
37        self.export_many([root])
38    }
39
40    /// Exports the DAGs rooted at `roots` as a single CARv1 archive sharing one block stream.
41    pub fn export_many(&self, roots: impl IntoIterator<Item = Cid>) -> CarExport {
42        CarExport {
43            dag: self.clone(),
44            roots: roots.into_iter().collect(),
45            fetch: false,
46            inner: None,
47        }
48    }
49
50    /// Imports a CAR archive from `reader`, storing every block and returning the archive roots.
51    pub fn import<R>(&self, reader: R) -> CarImport<R>
52    where
53        R: AsyncRead + Unpin + Send + 'static,
54    {
55        CarImport {
56            dag: self.clone(),
57            reader: Some(reader),
58            pin: false,
59            inner: None,
60        }
61    }
62}
63
64/// A CARv1 export
65pub struct CarExport {
66    dag: IpldDag,
67    roots: Vec<Cid>,
68    fetch: bool,
69    inner: Option<BoxStream<'static, Result<Bytes, Error>>>,
70}
71
72impl CarExport {
73    /// Fetches any block missing from the local store over the network.
74    pub fn fetch(mut self) -> Self {
75        self.fetch = true;
76        self
77    }
78
79    /// Writes the archive to `path`, streaming without buffering the whole thing.
80    #[cfg(not(target_arch = "wasm32"))]
81    pub async fn to_file(self, path: impl AsRef<Path>) -> Result<(), Error> {
82        let mut file = tokio::fs::File::create(path.as_ref()).await?;
83        let mut stream = export_stream(self.dag, self.roots, self.fetch);
84        while let Some(chunk) = stream.next().await {
85            file.write_all(&chunk?).await?;
86        }
87        file.flush().await?;
88        Ok(())
89    }
90
91    fn stream(&mut self) -> &mut BoxStream<'static, Result<Bytes, Error>> {
92        if self.inner.is_none() {
93            self.inner = Some(export_stream(
94                self.dag.clone(),
95                self.roots.clone(),
96                self.fetch,
97            ));
98        }
99        self.inner.as_mut().expect("just initialized")
100    }
101}
102
103impl Stream for CarExport {
104    type Item = Result<Bytes, Error>;
105
106    fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
107        self.get_mut().stream().poll_next_unpin(cx)
108    }
109}
110
111impl IntoFuture for CarExport {
112    type Output = Result<Bytes, Error>;
113    type IntoFuture = BoxFuture<'static, Result<Bytes, Error>>;
114
115    fn into_future(self) -> Self::IntoFuture {
116        async move {
117            let mut stream = export_stream(self.dag, self.roots, self.fetch);
118            let mut out = BytesMut::new();
119            while let Some(chunk) = stream.try_next().await? {
120                out.extend_from_slice(&chunk);
121            }
122            Ok(out.freeze())
123        }
124        .boxed()
125    }
126}
127
128fn export_stream(
129    dag: IpldDag,
130    roots: Vec<Cid>,
131    fetch: bool,
132) -> BoxStream<'static, Result<Bytes, Error>> {
133    async_stream::try_stream! {
134        if fetch && dag.ipfs.is_none() {
135            Err::<(), _>(anyhow!("export .fetch() requires an Ipfs-backed dag; this dag is repo-only"))?;
136        }
137
138        yield frame(&encode_v1_header(&roots)?);
139
140        let mut seen = HashSet::new();
141        let mut stack: Vec<Cid> = roots.iter().rev().copied().collect();
142        while let Some(cid) = stack.pop() {
143            if !seen.insert(cid) {
144                continue;
145            }
146            let block = export_block(&dag, cid, fetch).await?;
147
148            let mut payload = block.cid().to_bytes();
149            payload.extend_from_slice(block.data());
150            yield frame(&payload);
151
152            let mut refs = Vec::new();
153            // an unknown codec has no links we can read; archive the block as an opaque leaf
154            // rather than aborting the whole export
155            if let Err(e) = block.references(&mut refs)
156                && e.kind() != std::io::ErrorKind::Unsupported
157            {
158                Err::<(), Error>(e.into())?;
159            }
160            for child in refs.into_iter().rev() {
161                if !seen.contains(&child) {
162                    stack.push(child);
163                }
164            }
165        }
166    }
167    .boxed()
168}
169
170async fn export_block(dag: &IpldDag, cid: Cid, fetch: bool) -> Result<Block, Error> {
171    if fetch {
172        dag.repo
173            .get_block(cid)
174            .set_local(false)
175            .timeout(Some(FETCH_TIMEOUT))
176            .await
177    } else {
178        dag.repo
179            .get_block_now(cid)
180            .await?
181            .ok_or_else(|| anyhow!("missing block {cid}; not in the local store"))
182    }
183}
184
185#[derive(Debug)]
186pub enum ImportStatus {
187    Progress { blocks: usize, bytes: u64 },
188    Completed { roots: Vec<Cid>, blocks: usize },
189    Failed { error: Error },
190}
191
192pub struct CarImport<R> {
193    dag: IpldDag,
194    reader: Option<R>,
195    pin: bool,
196    inner: Option<BoxStream<'static, ImportStatus>>,
197}
198
199impl<R> CarImport<R>
200where
201    R: AsyncRead + Unpin + Send + 'static,
202{
203    /// Recursively pins each root after a successful import.
204    pub fn pin_roots(mut self) -> Self {
205        self.pin = true;
206        self
207    }
208
209    fn stream(&mut self) -> &mut BoxStream<'static, ImportStatus> {
210        if self.inner.is_none() {
211            let reader = self.reader.take().expect("CarImport already consumed");
212            self.inner = Some(import_status_stream(self.dag.clone(), reader, self.pin).boxed());
213        }
214        self.inner.as_mut().expect("just initialized")
215    }
216}
217
218impl<R> Stream for CarImport<R>
219where
220    R: AsyncRead + Unpin + Send + 'static,
221{
222    type Item = ImportStatus;
223
224    fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
225        self.get_mut().stream().poll_next_unpin(cx)
226    }
227}
228
229impl<R> IntoFuture for CarImport<R>
230where
231    R: AsyncRead + Unpin + Send + 'static,
232{
233    type Output = Result<Vec<Cid>, Error>;
234    type IntoFuture = BoxFuture<'static, Result<Vec<Cid>, Error>>;
235
236    fn into_future(self) -> Self::IntoFuture {
237        let CarImport {
238            dag, reader, pin, ..
239        } = self;
240        let reader = reader.expect("CarImport already consumed");
241        async move {
242            let stream = import_status_stream(dag, reader, pin);
243            futures::pin_mut!(stream);
244            let mut roots = None;
245            while let Some(status) = stream.next().await {
246                match status {
247                    ImportStatus::Completed { roots: done, .. } => roots = Some(done),
248                    ImportStatus::Failed { error } => return Err(error),
249                    ImportStatus::Progress { .. } => {}
250                }
251            }
252            roots.ok_or_else(|| anyhow!("import produced no result"))
253        }
254        .boxed()
255    }
256}
257
258fn import_status_stream<R>(dag: IpldDag, reader: R, pin: bool) -> impl Stream<Item = ImportStatus>
259where
260    R: AsyncRead + Unpin,
261{
262    async_stream::stream! {
263        let mut reader = BufReader::new(reader);
264        let (roots, limit) = match read_car_header(&mut reader).await {
265            Ok(value) => value,
266            Err(error) => {
267                yield ImportStatus::Failed { error };
268                return;
269            }
270        };
271
272        let mut block_reader = (&mut reader).take(limit);
273        let mut batch = Vec::with_capacity(IMPORT_BATCH);
274        let mut blocks = 0usize;
275        let mut bytes = 0u64;
276
277        loop {
278            let body = match read_frame(&mut block_reader, MAX_SECTION_LEN).await {
279                Ok(Some(body)) => body,
280                Ok(None) => break,
281                Err(error) => {
282                    yield ImportStatus::Failed { error };
283                    return;
284                }
285            };
286            let mut cursor = std::io::Cursor::new(&body[..]);
287            let cid = match Cid::read_bytes(&mut cursor) {
288                Ok(cid) => cid,
289                Err(e) => {
290                    yield ImportStatus::Failed { error: anyhow!("malformed cid in CAR: {e}") };
291                    return;
292                }
293            };
294            let data = body[cursor.position() as usize..].to_vec();
295            let block = match make_block(cid, data) {
296                Ok(block) => block,
297                Err(error) => {
298                    yield ImportStatus::Failed { error };
299                    return;
300                }
301            };
302            blocks += 1;
303            bytes += body.len() as u64;
304            batch.push(block);
305            if batch.len() >= IMPORT_BATCH {
306                if let Err(error) = dag.repo.put_blocks(std::mem::take(&mut batch)).await {
307                    yield ImportStatus::Failed { error };
308                    return;
309                }
310                yield ImportStatus::Progress { blocks, bytes };
311            }
312        }
313
314        if !batch.is_empty()
315            && let Err(error) = dag.repo.put_blocks(batch).await
316        {
317            yield ImportStatus::Failed { error };
318            return;
319        }
320
321        if limit != u64::MAX && block_reader.limit() != 0 {
322            yield ImportStatus::Failed {
323                error: anyhow!("CARv2: inner archive shorter than its declared data size"),
324            };
325            return;
326        }
327
328        if pin {
329            for root in &roots {
330                match dag.repo.is_pinned(root).await {
331                    Ok(true) => {}
332                    Ok(false) => {
333                        if let Err(error) = dag.repo.pin(*root).recursive().local().await {
334                            yield ImportStatus::Failed { error };
335                            return;
336                        }
337                    }
338                    Err(error) => {
339                        yield ImportStatus::Failed { error };
340                        return;
341                    }
342                }
343            }
344        }
345
346        yield ImportStatus::Progress { blocks, bytes };
347        yield ImportStatus::Completed { roots, blocks };
348    }
349}
350
351async fn read_car_header<R>(reader: &mut BufReader<R>) -> Result<(Vec<Cid>, u64), Error>
352where
353    R: AsyncRead + Unpin,
354{
355    let header = read_frame(reader, MAX_HEADER_LEN)
356        .await?
357        .ok_or_else(|| anyhow!("empty CAR: no header"))?;
358
359    match parse_header(&header)? {
360        Header::V1 { roots } => Ok((roots, u64::MAX)),
361        Header::V2Pragma => {
362            if header != CAR_V2_PRAGMA[1..] {
363                return Err(anyhow!("non-canonical CARv2 pragma"));
364            }
365            let v2 = read_exact_vec(reader, V2_HEADER_LEN).await?;
366            let data_offset = u64::from_le_bytes(v2[16..24].try_into().expect("8 bytes"));
367            let data_size = u64::from_le_bytes(v2[24..32].try_into().expect("8 bytes"));
368            let consumed = (CAR_V2_PRAGMA.len() + V2_HEADER_LEN) as u64;
369            if data_offset < consumed {
370                return Err(anyhow!(
371                    "malformed CARv2: data offset {data_offset} precedes the header"
372                ));
373            }
374            skip(reader, data_offset - consumed).await?;
375            let mut inner = (&mut *reader).take(data_size);
376            let inner_header = read_frame(&mut inner, MAX_HEADER_LEN)
377                .await?
378                .ok_or_else(|| anyhow!("CARv2: empty inner archive"))?;
379            let roots = match parse_header(&inner_header)? {
380                Header::V1 { roots } => roots,
381                Header::V2Pragma => return Err(anyhow!("nested CARv2 is not supported")),
382            };
383            Ok((roots, inner.limit()))
384        }
385    }
386}
387
388/// Builds a block from a CAR section. Blocks whose multihash this build can recompute are verified
389/// (rejecting a corrupt archive), an identity (0x00) block is checked against its data directly, and
390/// any other hash function is trusted as the `(cid, bytes)` pairing.
391fn make_block(cid: Cid, data: Vec<u8>) -> Result<Block, Error> {
392    let code = cid.hash().code();
393    if code == 0x00 {
394        if cid.hash().digest() != data.as_slice() {
395            return Err(anyhow!("identity block {cid} does not match its data"));
396        }
397        return Ok(Block::new_unchecked(cid, data));
398    }
399    if multihash_codetable::Code::try_from(code).is_err() {
400        return Ok(Block::new_unchecked(cid, data));
401    }
402    Block::new(cid, data).map_err(|e| anyhow!("invalid block {cid} in CAR: {e}"))
403}
404
405enum Header {
406    V1 { roots: Vec<Cid> },
407    V2Pragma,
408}
409
410fn parse_header(bytes: &[u8]) -> Result<Header, Error> {
411    let ipld: Ipld = serde_ipld_dagcbor::codec::DagCborCodec::decode_from_slice(bytes)
412        .map_err(|e| anyhow!("invalid CAR header: {e}"))?;
413    let Ipld::Map(map) = ipld else {
414        return Err(anyhow!("CAR header is not a map"));
415    };
416    let version = match map.get("version") {
417        Some(Ipld::Integer(v)) => *v,
418        _ => return Err(anyhow!("CAR header has no version")),
419    };
420    match version {
421        1 => {
422            let roots = match map.get("roots") {
423                Some(Ipld::List(list)) => list
424                    .iter()
425                    .map(|item| match item {
426                        Ipld::Link(cid) => Ok(*cid),
427                        _ => Err(anyhow!("CAR root is not a cid")),
428                    })
429                    .collect::<Result<Vec<_>, _>>()?,
430                Some(_) => return Err(anyhow!("CAR roots is not a list")),
431                None => Vec::new(),
432            };
433            Ok(Header::V1 { roots })
434        }
435        2 => Ok(Header::V2Pragma),
436        other => Err(anyhow!("unsupported CAR version {other}")),
437    }
438}
439
440fn encode_v1_header(roots: &[Cid]) -> Result<Vec<u8>, Error> {
441    let mut map = BTreeMap::new();
442    map.insert(
443        "roots".to_string(),
444        Ipld::List(roots.iter().map(|c| Ipld::Link(*c)).collect()),
445    );
446    map.insert("version".to_string(), Ipld::Integer(1));
447    serde_ipld_dagcbor::codec::DagCborCodec::encode_to_vec(&Ipld::Map(map))
448        .map_err(|e| anyhow!("encoding CAR header failed: {e}"))
449}
450
451fn frame(payload: &[u8]) -> Bytes {
452    let mut buf = Vec::with_capacity(payload.len() + 10);
453    let mut varint = unsigned_varint::encode::u64_buffer();
454    buf.extend_from_slice(unsigned_varint::encode::u64(
455        payload.len() as u64,
456        &mut varint,
457    ));
458    buf.extend_from_slice(payload);
459    Bytes::from(buf)
460}
461
462async fn read_uvarint<R>(reader: &mut R) -> Result<Option<u64>, Error>
463where
464    R: AsyncRead + Unpin,
465{
466    let mut value = 0u64;
467    let mut shift = 0u32;
468    loop {
469        let mut byte = [0u8; 1];
470        if reader.read(&mut byte).await? == 0 {
471            if shift == 0 {
472                return Ok(None);
473            }
474            return Err(anyhow!("unexpected end of CAR in a varint"));
475        }
476        // the 10th byte (shift 63) may only carry the single top bit, with no continuation
477        if shift == 63 && byte[0] > 0x01 {
478            return Err(anyhow!("CAR varint overflows u64"));
479        }
480        value |= u64::from(byte[0] & 0x7f) << shift;
481        if byte[0] & 0x80 == 0 {
482            return Ok(Some(value));
483        }
484        shift += 7;
485        if shift >= 64 {
486            return Err(anyhow!("CAR varint overflows u64"));
487        }
488    }
489}
490
491async fn read_exact_vec<R>(reader: &mut R, n: usize) -> Result<Vec<u8>, Error>
492where
493    R: AsyncRead + Unpin,
494{
495    let mut buf = vec![0u8; n];
496    reader.read_exact(&mut buf).await?;
497    Ok(buf)
498}
499
500/// Reads one length-prefixed frame, capping the declared length at `max` before allocating.
501/// Returns `None` at a clean end-of-stream boundary.
502async fn read_frame<R>(reader: &mut R, max: u64) -> Result<Option<Vec<u8>>, Error>
503where
504    R: AsyncRead + Unpin,
505{
506    let Some(len) = read_uvarint(reader).await? else {
507        return Ok(None);
508    };
509    if len > max {
510        return Err(anyhow!(
511            "CAR frame length {len} exceeds the {max}-byte limit"
512        ));
513    }
514    Ok(Some(read_exact_vec(reader, len as usize).await?))
515}
516
517async fn skip<R>(reader: &mut R, mut n: u64) -> Result<(), Error>
518where
519    R: AsyncRead + Unpin,
520{
521    let mut scratch = [0u8; 4096];
522    while n > 0 {
523        let take = n.min(scratch.len() as u64) as usize;
524        reader.read_exact(&mut scratch[..take]).await?;
525        n -= take as u64;
526    }
527    Ok(())
528}
529
530#[cfg(test)]
531mod tests {
532    use super::*;
533    use crate::repo::Repo;
534
535    async fn new_dag() -> (Repo<crate::repo::DefaultStorage>, IpldDag) {
536        let repo = Repo::new_memory();
537        repo.init().await.unwrap();
538        let dag = IpldDag::from(repo.clone());
539        (repo, dag)
540    }
541
542    async fn sample_dag(dag: &IpldDag) -> (Cid, Cid) {
543        let leaf = dag.put_dag("hello car").await.unwrap();
544        let mut map = BTreeMap::new();
545        map.insert("child".to_string(), Ipld::Link(leaf));
546        let root = dag.put_dag(Ipld::Map(map)).await.unwrap();
547        (root, leaf)
548    }
549
550    #[tokio::test]
551    async fn export_import_roundtrip() {
552        let (_repo, dag) = new_dag().await;
553        let (root, leaf) = sample_dag(&dag).await;
554
555        let car = dag.export(root).await.unwrap();
556        assert!(!car.is_empty());
557
558        let (dest_repo, dest) = new_dag().await;
559        let roots = dest
560            .import(futures::io::Cursor::new(car.clone()))
561            .await
562            .unwrap();
563        assert_eq!(roots, vec![root]);
564
565        assert!(dest_repo.contains(&root).await.unwrap());
566        assert!(dest_repo.contains(&leaf).await.unwrap());
567        assert_eq!(
568            dest.get_dag(leaf).await.unwrap(),
569            Ipld::String("hello car".into())
570        );
571    }
572
573    #[tokio::test]
574    async fn streamed_export_matches_collected() {
575        let (_repo, dag) = new_dag().await;
576        let (root, _) = sample_dag(&dag).await;
577
578        let collected = dag.export(root).await.unwrap();
579
580        let mut streamed = Vec::new();
581        let export = dag.export(root);
582        futures::pin_mut!(export);
583        while let Some(chunk) = export.next().await {
584            streamed.extend_from_slice(&chunk.unwrap());
585        }
586        assert_eq!(streamed, collected);
587    }
588
589    #[tokio::test]
590    async fn export_missing_block_errors() {
591        let (_repo, dag) = new_dag().await;
592        let absent =
593            Cid::try_from("bafyreib2rxk3rybk3aobmv5cjuql3bm2twh4jo5uxgt5e7ym74n5lvrxle").unwrap();
594        assert!(dag.export(absent).await.is_err());
595    }
596
597    #[tokio::test]
598    async fn import_pins_roots_when_requested() {
599        let (_repo, dag) = new_dag().await;
600        let (root, leaf) = sample_dag(&dag).await;
601        let car = dag.export(root).await.unwrap();
602
603        let (dest_repo, dest) = new_dag().await;
604        dest.import(futures::io::Cursor::new(car.clone()))
605            .pin_roots()
606            .await
607            .unwrap();
608        assert!(dest_repo.is_pinned(&root).await.unwrap());
609        assert!(
610            dest_repo.is_pinned(&leaf).await.unwrap(),
611            "recursive pin reaches children"
612        );
613    }
614
615    #[tokio::test]
616    async fn imports_carv2_wrapping_a_v1_payload() {
617        let (_repo, dag) = new_dag().await;
618        let (root, leaf) = sample_dag(&dag).await;
619        let v1 = dag.export(root).await.unwrap();
620
621        // wrap the v1 payload in a minimal CARv2: pragma, 40-byte header, then the payload (no index)
622        let mut v2 = Vec::new();
623        v2.extend_from_slice(&CAR_V2_PRAGMA);
624        v2.extend_from_slice(&[0u8; 16]); // characteristics
625        let data_offset = (CAR_V2_PRAGMA.len() + V2_HEADER_LEN) as u64;
626        v2.extend_from_slice(&data_offset.to_le_bytes());
627        v2.extend_from_slice(&(v1.len() as u64).to_le_bytes());
628        v2.extend_from_slice(&0u64.to_le_bytes()); // index offset (none)
629        v2.extend_from_slice(&v1);
630
631        let (dest_repo, dest) = new_dag().await;
632        let roots = dest
633            .import(futures::io::Cursor::new(v2.clone()))
634            .await
635            .unwrap();
636        assert_eq!(roots, vec![root]);
637        assert!(dest_repo.contains(&leaf).await.unwrap());
638    }
639
640    #[tokio::test]
641    async fn import_rejects_a_corrupt_block() {
642        let (_repo, dag) = new_dag().await;
643        let (root, _) = sample_dag(&dag).await;
644        let mut car = dag.export(root).await.unwrap().to_vec();
645        // flip a byte in the last block's payload so its cid no longer matches its data
646        *car.last_mut().unwrap() ^= 0xff;
647
648        let (_dest_repo, dest) = new_dag().await;
649        assert!(
650            dest.import(futures::io::Cursor::new(car.clone()))
651                .await
652                .is_err()
653        );
654    }
655
656    #[tokio::test]
657    async fn exports_unknown_codec_block_as_leaf() {
658        use multihash_codetable::{Code, MultihashDigest};
659        let (repo, dag) = new_dag().await;
660        let opaque = b"bytes under a codec we cannot decode";
661        let unk = Cid::new_v1(0x0200, Code::Sha2_256.digest(opaque));
662        repo.put_block(&Block::new(unk, opaque.to_vec()).unwrap())
663            .await
664            .unwrap();
665
666        let mut map = BTreeMap::new();
667        map.insert("opaque".to_string(), Ipld::Link(unk));
668        let root = dag.put_dag(Ipld::Map(map)).await.unwrap();
669
670        // export must archive the undecodable child as a leaf, not abort
671        let car = dag.export(root).await.unwrap();
672        let (dest_repo, dest) = new_dag().await;
673        let roots = dest.import(futures::io::Cursor::new(car)).await.unwrap();
674        assert_eq!(roots, vec![root]);
675        assert!(dest_repo.contains(&unk).await.unwrap());
676    }
677
678    #[tokio::test]
679    async fn imports_identity_hashed_block() {
680        use ipld_core::cid::multihash::Multihash;
681        let data = b"inline identity payload";
682        let cid = Cid::new_v1(0x55, Multihash::wrap(0x00, data).unwrap());
683
684        let mut payload = cid.to_bytes();
685        payload.extend_from_slice(data);
686        let mut car = Vec::new();
687        car.extend_from_slice(&frame(&encode_v1_header(&[cid]).unwrap()));
688        car.extend_from_slice(&frame(&payload));
689
690        let (dest_repo, dest) = new_dag().await;
691        let roots = dest.import(futures::io::Cursor::new(car)).await.unwrap();
692        assert_eq!(roots, vec![cid]);
693        assert!(dest_repo.contains(&cid).await.unwrap());
694    }
695
696    #[tokio::test]
697    async fn import_rejects_oversized_frame() {
698        let mut car = Vec::new();
699        car.extend_from_slice(&frame(&encode_v1_header(&[]).unwrap()));
700        let mut vbuf = unsigned_varint::encode::u64_buffer();
701        car.extend_from_slice(unsigned_varint::encode::u64(MAX_SECTION_LEN + 1, &mut vbuf));
702
703        let (_r, dest) = new_dag().await;
704        assert!(dest.import(futures::io::Cursor::new(car)).await.is_err());
705    }
706
707    #[tokio::test]
708    async fn pin_roots_is_idempotent_across_imports() {
709        let (_repo, dag) = new_dag().await;
710        let (root, _) = sample_dag(&dag).await;
711        let car = dag.export(root).await.unwrap();
712
713        let (dest_repo, dest) = new_dag().await;
714        dest.import(futures::io::Cursor::new(car.clone()))
715            .pin_roots()
716            .await
717            .unwrap();
718        dest.import(futures::io::Cursor::new(car.clone()))
719            .pin_roots()
720            .await
721            .unwrap();
722        assert!(dest_repo.is_pinned(&root).await.unwrap());
723    }
724
725    #[tokio::test]
726    async fn import_reports_progress() {
727        let (_repo, dag) = new_dag().await;
728        let mut links = BTreeMap::new();
729        for i in 0..300u32 {
730            let leaf = dag.put_dag(format!("leaf {i}")).await.unwrap();
731            links.insert(format!("l{i:03}"), Ipld::Link(leaf));
732        }
733        let root = dag.put_dag(Ipld::Map(links)).await.unwrap();
734        let car = dag.export(root).await.unwrap();
735
736        let (dest_repo, dest) = new_dag().await;
737        let mut import = dest.import(futures::io::Cursor::new(car));
738        let mut last = 0usize;
739        let mut progress_events = 0usize;
740        let mut completed = None;
741        while let Some(status) = import.next().await {
742            match status {
743                ImportStatus::Progress { blocks, bytes } => {
744                    assert!(blocks >= last);
745                    assert!(bytes > 0);
746                    last = blocks;
747                    progress_events += 1;
748                }
749                ImportStatus::Completed { roots, blocks } => completed = Some((roots, blocks)),
750                ImportStatus::Failed { error } => panic!("import failed: {error}"),
751            }
752        }
753
754        let (roots, blocks) = completed.unwrap();
755        assert_eq!(roots, vec![root]);
756        assert_eq!(blocks, 301);
757        assert_eq!(last, 301);
758        assert!(
759            progress_events >= 2,
760            "batched flush plus a final progress, got {progress_events}"
761        );
762        assert!(dest_repo.contains(&root).await.unwrap());
763    }
764}
rust_ipfs/dag/car.rs

rust_ipfs/dag/
car.rs
