feat: implement dag import/export (#3728)

Adds `ipfs.dag.import` and `ipfs.dag.export` commands to import/export CAR files,
e.g. single-file archives that contain blocks and root CIDs.

Supersedes #2953
Fixes #2745

Co-authored-by: achingbrain <alex@achingbrain.net>

Author: rvagg

docs/core-api/DAG.md

@@ -27,6 +27,46 @@ _Explore the DAG API through interactive coding challenges in our ProtoSchool tu
 - _[P2P data links with content addressing](https://proto.school/#/basics/) (beginner)_
 - _[Blogging on the Decentralized Web](https://proto.school/#/blog/) (intermediate)_
 
+## `ipfs.dag.export(cid, [options])`
+
+> Returns a stream of Uint8Arrays that make up a [CAR file][]
+
+Exports a CAR for the entire DAG available from the given root CID. The CAR will have a single
+root and IPFS will attempt to fetch and bundle all blocks that are linked within the connected
+DAG.
+
+### Parameters
+
+| Name | Type | Description |
+| ---- | ---- | ----------- |
+| cid  | [CID][] | The root CID of the DAG we wish to export |
+
+### Options
+
+An optional object which may have the following keys:
+
+| Name | Type | Default | Description |
+| ---- | ---- | ------- | ----------- |
+| timeout | `Number` | `undefined` | A timeout in ms |
+| signal | [AbortSignal][] | `undefined` |  Can be used to cancel any long running requests started as a result of this call |
+
+### Returns
+
+| Type | Description |
+| -------- | -------- |
+| `AsyncIterable<Uint8Array>` | A stream containing the car file bytes |
+
+### Example
+
+```JavaScript
+const { Readable } = require('stream')
+
+const out = await ipfs.dag.export(cid)
+
+Readable.from(out).pipe(fs.createWriteStream('example.car'))
+```
+
+A great source of [examples][] can be found in the tests for this API.
 ## `ipfs.dag.put(dagNode, [options])`
 
 > Store an IPLD format node
@@ -146,6 +186,48 @@ await getAndLog(cid, '/c/ca/1')
 
 A great source of [examples][] can be found in the tests for this API.
 
+## `ipfs.dag.import(source, [options])`
+
+> Adds one or more [CAR file][]s full of blocks to the repo for this node
+
+Import all blocks from one or more CARs and optionally recursively pin the roots identified
+within the CARs.
+
+### Parameters
+
+| Name | Type | Description |
+| ---- | ---- | ----------- |
+| sources | `AsyncIterable<Uint8Array>` | `AsyncIterable<AsyncIterable<Uint8Array>>` | One or more [CAR file][] streams |
+
+### Options
+
+An optional object which may have the following keys:
+
+| Name | Type | Default | Description |
+| ---- | ---- | ------- | ----------- |
+| pinRoots | `boolean` | `true` | Whether to recursively pin each root to the blockstore |
+| timeout | `Number` | `undefined` | A timeout in ms |
+| signal | [AbortSignal][] | `undefined` |  Can be used to cancel any long running requests started as a result of this call |
+
+### Returns
+
+| Type | Description |
+| -------- | -------- |
+| `AsyncIterable<{ cid: CID, pinErrorMsg?: string }>` | A stream containing the result of importing the car file(s) |
+
+### Example
+
+```JavaScript
+const fs = require('fs')
+
+for await (const result of ipfs.dag.import(fs.createReadStream('./path/to/archive.car'))) {
+  console.info(result)
+  // Qmfoo
+}
+```
+
+A great source of [examples][] can be found in the tests for this API.
+
 ## `ipfs.dag.tree(cid, [options])`
 
 > Enumerate all the entries in a graph
@@ -262,7 +344,7 @@ console.log(result)
 
 A great source of [examples][] can be found in the tests for this API.
 
-
 [examples]: https://github.com/ipfs/js-ipfs/blob/master/packages/interface-ipfs-core/src/dag
 [cid]: https://www.npmjs.com/package/cids
 [AbortSignal]: https://developer.mozilla.org/en-US/docs/Web/API/AbortSignal
+[CAR file]: https://ipld.io/specs/transport/car/

examples/custom-ipfs-repo/package.json

@@ -13,9 +13,9 @@
     "@ipld/dag-cbor": "^6.0.5",
     "@ipld/dag-pb": "^2.1.3",
     "blockstore-datastore-adapter": "^1.0.0",
-    "datastore-fs": "^5.0.1",
+    "datastore-fs": "^5.0.2",
     "ipfs": "^0.55.4",
-    "ipfs-repo": "^11.0.0",
+    "ipfs-repo": "^11.0.1",
     "it-all": "^1.0.4",
     "multiformats": "^9.4.1"
   },

packages/interface-ipfs-core/package.json

@@ -36,6 +36,7 @@
     ]
   },
   "dependencies": {
+    "@ipld/car": "^3.1.6",
     "@ipld/dag-cbor": "^6.0.5",
     "@ipld/dag-pb": "^2.1.3",
     "abort-controller": "^3.0.0",
@@ -56,7 +57,8 @@
     "it-first": "^1.0.4",
     "it-last": "^1.0.4",
     "it-map": "^1.0.4",
-    "it-pushable": "^1.4.0",
+    "it-pushable": "^1.4.2",
+    "it-to-buffer": "^2.0.0",
     "libp2p-crypto": "^0.19.6",
     "libp2p-websockets": "^0.16.1",
     "multiaddr": "^10.0.0",

packages/interface-ipfs-core/src/dag/export.js

@@ -0,0 +1,90 @@
+/* eslint-env mocha */
+'use strict'
+
+const all = require('it-all')
+const { getDescribe, getIt, expect } = require('../utils/mocha')
+const { CarReader } = require('@ipld/car')
+const uint8ArrayFromString = require('uint8arrays/from-string')
+const dagPb = require('@ipld/dag-pb')
+const dagCbor = require('@ipld/dag-cbor')
+const loadFixture = require('aegir/utils/fixtures')
+const toBuffer = require('it-to-buffer')
+
+/** @typedef { import("ipfsd-ctl/src/factory") } Factory */
+/**
+ * @param {Factory} common
+ * @param {Object} options
+ */
+module.exports = (common, options) => {
+  const describe = getDescribe(options)
+  const it = getIt(options)
+
+  describe('.dag.export', () => {
+    let ipfs
+    before(async () => {
+      ipfs = (await common.spawn()).api
+    })
+
+    after(() => common.clean())
+
+    it('should export a car file', async () => {
+      const child = dagPb.encode({
+        Data: uint8ArrayFromString('block-' + Math.random()),
+        Links: []
+      })
+      const childCid = await ipfs.block.put(child, {
+        format: 'dag-pb',
+        version: 0
+      })
+      const parent = dagPb.encode({
+        Links: [{
+          Hash: childCid,
+          Tsize: child.length,
+          Name: ''
+        }]
+      })
+      const parentCid = await ipfs.block.put(parent, {
+        format: 'dag-pb',
+        version: 0
+      })
+      const grandParent = dagCbor.encode({
+        parent: parentCid
+      })
+      const grandParentCid = await await ipfs.block.put(grandParent, {
+        format: 'dag-cbor',
+        version: 1
+      })
+
+      const expectedCids = [
+        grandParentCid,
+        parentCid,
+        childCid
+      ]
+
+      const reader = await CarReader.fromIterable(ipfs.dag.export(grandParentCid))
+      const cids = await all(reader.cids())
+
+      expect(cids).to.deep.equal(expectedCids)
+    })
+
+    it('export of shuffled devnet export identical to canonical original', async function () {
+      this.timeout(360000)
+
+      const input = loadFixture('test/fixtures/car/lotus_devnet_genesis.car', 'interface-ipfs-core')
+      const result = await all(ipfs.dag.import(async function * () { yield input }()))
+      const exported = await toBuffer(ipfs.dag.export(result[0].root.cid))
+
+      expect(exported).to.equalBytes(input)
+    })
+
+    it('export of shuffled testnet export identical to canonical original', async function () {
+      this.timeout(360000)
+
+      const input = loadFixture('test/fixtures/car/lotus_testnet_export_128.car', 'interface-ipfs-core')
+      const result = await all(ipfs.dag.import(async function * () { yield input }()))
+      const exported = await toBuffer(ipfs.dag.export(result[0].root.cid))
+
+      expect(exported).to.equalBytes(input)
+    })
+  })
+}

packages/interface-ipfs-core/src/dag/import.js

@@ -0,0 +1,156 @@
+/* eslint-env mocha */
+'use strict'
+
+const all = require('it-all')
+const drain = require('it-drain')
+const { CID } = require('multiformats/cid')
+const { sha256 } = require('multiformats/hashes/sha2')
+const { getDescribe, getIt, expect } = require('../utils/mocha')
+const { CarWriter, CarReader } = require('@ipld/car')
+const raw = require('multiformats/codecs/raw')
+const uint8ArrayFromString = require('uint8arrays/from-string')
+const loadFixture = require('aegir/utils/fixtures')
+
+/**
+ *
+ * @param {number} num
+ */
+async function createBlocks (num) {
+  const blocks = []
+
+  for (let i = 0; i < num; i++) {
+    const bytes = uint8ArrayFromString('block-' + Math.random())
+    const digest = await sha256.digest(raw.encode(bytes))
+    const cid = CID.create(1, raw.code, digest)
+
+    blocks.push({ bytes, cid })
+  }
+
+  return blocks
+}
+
+/**
+ * @param {{ cid: CID, bytes: Uint8Array }[]} blocks
+ * @returns {AsyncIterable<Uint8Array>}
+ */
+async function createCar (blocks) {
+  const rootBlock = blocks[0]
+  const { writer, out } = await CarWriter.create([rootBlock.cid])
+
+  writer.put(rootBlock)
+    .then(async () => {
+      for (const block of blocks.slice(1)) {
+        writer.put(block)
+      }
+
+      await writer.close()
+    })
+
+  return out
+}
+
+/** @typedef { import("ipfsd-ctl/src/factory") } Factory */
+/**
+ * @param {Factory} common
+ * @param {Object} options
+ */
+module.exports = (common, options) => {
+  const describe = getDescribe(options)
+  const it = getIt(options)
+
+  describe('.dag.import', () => {
+    let ipfs
+    before(async () => {
+      ipfs = (await common.spawn()).api
+    })
+
+    after(() => common.clean())
+
+    it('should import a car file', async () => {
+      const blocks = await createBlocks(5)
+      const car = await createCar(blocks)
+
+      const result = await all(ipfs.dag.import(car))
+      expect(result).to.have.lengthOf(1)
+      expect(result).to.have.nested.deep.property('[0].root.cid', blocks[0].cid)
+
+      for (const { cid } of blocks) {
+        await expect(ipfs.block.get(cid)).to.eventually.be.ok()
+      }
+
+      await expect(all(ipfs.pin.ls({ paths: blocks[0].cid }))).to.eventually.have.lengthOf(1)
+        .and.have.nested.property('[0].type', 'recursive')
+    })
+
+    it('should import a car file without pinning the roots', async () => {
+      const blocks = await createBlocks(5)
+      const car = await createCar(blocks)
+
+      await all(ipfs.dag.import(car, {
+        pinRoots: false
+      }))
+
+      await expect(all(ipfs.pin.ls({ paths: blocks[0].cid }))).to.eventually.be.rejectedWith(/is not pinned/)
+    })
+
+    it('should import multiple car files', async () => {
+      const blocks1 = await createBlocks(5)
+      const car1 = await createCar(blocks1)
+
+      const blocks2 = await createBlocks(5)
+      const car2 = await createCar(blocks2)
+
+      const result = await all(ipfs.dag.import([car1, car2]))
+      expect(result).to.have.lengthOf(2)
+      expect(result).to.deep.include({ root: { cid: blocks1[0].cid, pinErrorMsg: '' } })
+      expect(result).to.deep.include({ root: { cid: blocks2[0].cid, pinErrorMsg: '' } })
+
+      for (const { cid } of blocks1) {
+        await expect(ipfs.block.get(cid)).to.eventually.be.ok()
+      }
+
+      for (const { cid } of blocks2) {
+        await expect(ipfs.block.get(cid)).to.eventually.be.ok()
+      }
+    })
+
+    it('should import car with roots but no blocks', async () => {
+      const input = loadFixture('test/fixtures/car/combined_naked_roots_genesis_and_128.car', 'interface-ipfs-core')
+      const reader = await CarReader.fromBytes(input)
+      const cids = await reader.getRoots()
+
+      expect(cids).to.have.lengthOf(2)
+
+      // naked roots car does not contain blocks
+      const result1 = await all(ipfs.dag.import(async function * () { yield input }()))
+      expect(result1).to.deep.include({ root: { cid: cids[0], pinErrorMsg: 'blockstore: block not found' } })
+      expect(result1).to.deep.include({ root: { cid: cids[1], pinErrorMsg: 'blockstore: block not found' } })
+
+      await drain(ipfs.dag.import(async function * () { yield loadFixture('test/fixtures/car/lotus_devnet_genesis_shuffled_nulroot.car', 'interface-ipfs-core') }()))
+
+      // have some of the blocks now, should be able to pin one root
+      const result2 = await all(ipfs.dag.import(async function * () { yield input }()))
+      expect(result2).to.deep.include({ root: { cid: cids[0], pinErrorMsg: '' } })
+      expect(result2).to.deep.include({ root: { cid: cids[1], pinErrorMsg: 'blockstore: block not found' } })
+
+      await drain(ipfs.dag.import(async function * () { yield loadFixture('test/fixtures/car/lotus_testnet_export_128.car', 'interface-ipfs-core') }()))
+
+      // have all of the blocks now, should be able to pin both
+      const result3 = await all(ipfs.dag.import(async function * () { yield input }()))
+      expect(result3).to.deep.include({ root: { cid: cids[0], pinErrorMsg: '' } })
+      expect(result3).to.deep.include({ root: { cid: cids[1], pinErrorMsg: '' } })
+    })
+
+    it('should import lotus devnet genesis shuffled nulroot', async () => {
+      const input = loadFixture('test/fixtures/car/lotus_devnet_genesis_shuffled_nulroot.car', 'interface-ipfs-core')
+      const reader = await CarReader.fromBytes(input)
+      const cids = await reader.getRoots()
+
+      expect(cids).to.have.lengthOf(1)
+      expect(cids[0].toString()).to.equal('bafkqaaa')
+
+      const result = await all(ipfs.dag.import(async function * () { yield input }()))
+      expect(result).to.have.nested.deep.property('[0].root.cid', cids[0])
+    })
+  })
+}

packages/interface-ipfs-core/src/dag/index.js

@@ -2,8 +2,10 @@
 const { createSuite } = require('../utils/suite')
 
 const tests = {
+  export: require('./export'),
   get: require('./get'),
   put: require('./put'),
+  import: require('./import'),
   resolve: require('./resolve')
 }