feat: support aborting exports (#47)

* feat: support aborting exports

Adds a `signal` option to the exporter that will be passed to `ipld.get`,
allowing it to be notified that the user is no longer interested in
exporting the file/directory/etc.

Current behaviour is that if the node is offline and a block is not in
the repo is will throw, if it's online the CID will be added to the
bitswap want list via the block service.

Follow up PRs to ipld, block service and bitswap will impelement the
removal logic.

* docs: add signal option to readme

Author: achingbrain

packages/ipfs-unixfs-exporter/README.md

@@ -21,7 +21,7 @@
   - [Usage](#usage)
     - [Example](#example)
       - [API](#api)
-    - [`exporter(cid, ipld)`](#exportercid-ipld)
+    - [`exporter(cid, ipld, options)`](#exportercid-ipld-options)
       - [UnixFS V1 entries](#unixfs-v1-entries)
       - [Raw entries](#raw-entries)
       - [CBOR entries](#cbor-entries)
@@ -85,12 +85,16 @@ console.info(content) // 0, 1, 2, 3
 const exporter = require('ipfs-unixfs-exporter')
 ```
 
-### `exporter(cid, ipld)`
+### `exporter(cid, ipld, options)`
 
-Uses the given [js-ipld instance][] to fetch an IPFS node by it's CID.
+Uses the given [ipld](https://github.com/ipld/js-ipld) instance to fetch an IPFS node by it's CID.
 
 Returns a Promise which resolves to an `entry`.
 
+`options` is an optional object argument that might include the following keys:
+
+- `signal` ([AbortSignal](https://developer.mozilla.org/en-US/docs/Web/API/AbortSignal)): Used to cancel any network requests that are initiated as a result of this export
+
 #### UnixFS V1 entries
 
 Entries with a `dag-pb` codec `CID` return UnixFS V1 entries:

packages/ipfs-unixfs-exporter/package.json

@@ -35,11 +35,13 @@
   },
   "homepage": "https://github.com/ipfs/js-ipfs-unixfs#readme",
   "devDependencies": {
+    "abort-controller": "^3.0.0",
     "aegir": "^21.3.0",
     "async-iterator-all": "^1.0.0",
     "async-iterator-buffer-stream": "^1.0.0",
     "async-iterator-first": "^1.0.0",
     "chai": "^4.2.0",
+    "chai-as-promised": "^7.1.1",
     "detect-node": "^2.0.4",
     "dirty-chai": "^2.0.1",
     "ipfs-unixfs-importer": "^1.0.3",

packages/ipfs-unixfs-exporter/src/index.js

@@ -44,7 +44,7 @@ const cidAndRest = (path) => {
   throw errCode(new Error(`Unknown path type ${path}`), 'ERR_BAD_PATH')
 }
 
-const walkPath = async function * (path, ipld) {
+const walkPath = async function * (path, ipld, options) {
   let {
     cid,
     toResolve
@@ -54,7 +54,7 @@ const walkPath = async function * (path, ipld) {
   const startingDepth = toResolve.length
 
   while (true) {
-    const result = await resolve(cid, name, entryPath, toResolve, startingDepth, ipld)
+    const result = await resolve(cid, name, entryPath, toResolve, startingDepth, ipld, options)
 
     if (!result.entry && !result.next) {
       throw errCode(new Error(`Could not resolve ${path}`), 'ERR_NOT_FOUND')
@@ -76,27 +76,27 @@ const walkPath = async function * (path, ipld) {
   }
 }
 
-const exporter = (path, ipld) => {
-  return last(walkPath(path, ipld))
+const exporter = (path, ipld, options) => {
+  return last(walkPath(path, ipld, options))
 }
 
-const recursive = async function * (path, ipld) {
-  const node = await exporter(path, ipld)
+const recursive = async function * (path, ipld, options) {
+  const node = await exporter(path, ipld, options)
 
   yield node
 
   if (node.unixfs && node.unixfs.type.includes('dir')) {
-    for await (const child of recurse(node)) {
+    for await (const child of recurse(node, options)) {
       yield child
     }
   }
 
-  async function * recurse (node) {
-    for await (const file of node.content()) {
+  async function * recurse (node, options) {
+    for await (const file of node.content(options)) {
       yield file
 
       if (file.unixfs.type.includes('dir')) {
-        for await (const subFile of recurse(file)) {
+        for await (const subFile of recurse(file, options)) {
           yield subFile
         }
       }

packages/ipfs-unixfs-exporter/src/resolvers/dag-cbor.js

@@ -3,8 +3,8 @@
 const CID = require('cids')
 const errCode = require('err-code')
 
-const resolve = async (cid, name, path, toResolve, resolve, depth, ipld) => {
-  const node = await ipld.get(cid)
+const resolve = async (cid, name, path, toResolve, resolve, depth, ipld, options) => {
+  const node = await ipld.get(cid, options)
   let subObject = node
   let subPath = path
 

packages/ipfs-unixfs-exporter/src/resolvers/identity.js

@@ -16,7 +16,7 @@ const rawContent = (node) => {
   }
 }
 
-const resolve = async (cid, name, path, toResolve, resolve, depth, ipld) => {
+const resolve = async (cid, name, path, toResolve, resolve, depth, ipld, options) => {
   if (toResolve.length) {
     throw errCode(new Error(`No link named ${path} found in raw node ${cid.toBaseEncodedString()}`), 'ERR_NOT_FOUND')
   }

packages/ipfs-unixfs-exporter/src/resolvers/index.js

@@ -9,14 +9,14 @@ const resolvers = {
   identity: require('./identity')
 }
 
-const resolve = (cid, name, path, toResolve, depth, ipld) => {
+const resolve = (cid, name, path, toResolve, depth, ipld, options) => {
   const resolver = resolvers[cid.codec]
 
   if (!resolver) {
     throw errCode(new Error(`No resolver for codec ${cid.codec}`), 'ERR_NO_RESOLVER')
   }
 
-  return resolver(cid, name, path, toResolve, resolve, depth, ipld)
+  return resolver(cid, name, path, toResolve, resolve, depth, ipld, options)
 }
 
 module.exports = resolve

packages/ipfs-unixfs-exporter/src/resolvers/raw.js

@@ -15,12 +15,12 @@ const rawContent = (node) => {
   }
 }
 
-const resolve = async (cid, name, path, toResolve, resolve, depth, ipld) => {
+const resolve = async (cid, name, path, toResolve, resolve, depth, ipld, options) => {
   if (toResolve.length) {
     throw errCode(new Error(`No link named ${path} found in raw node ${cid.toBaseEncodedString()}`), 'ERR_NOT_FOUND')
   }
 
-  const buf = await ipld.get(cid)
+  const buf = await ipld.get(cid, options)
 
   return {
     entry: {

packages/ipfs-unixfs-exporter/src/resolvers/unixfs-v1/content/directory.js

@@ -1,13 +1,13 @@
 'use strict'
 
-const directoryContent = (cid, node, unixfs, path, resolve, depth, ipld) => {
+const directoryContent = (cid, node, unixfs, path, resolve, depth, ipld, options) => {
   return async function * (options = {}) {
     const offset = options.offset || 0
     const length = options.length || node.Links.length
     const links = node.Links.slice(offset, length)
 
     for (const link of links) {
-      const result = await resolve(link.Hash, link.Name, `${path}/${link.Name}`, [], depth + 1, ipld)
+      const result = await resolve(link.Hash, link.Name, `${path}/${link.Name}`, [], depth + 1, ipld, options)
 
       yield result.entry
     }

packages/ipfs-unixfs-exporter/src/resolvers/unixfs-v1/content/file.js

@@ -5,7 +5,7 @@ const validateOffsetAndLength = require('../../../utils/validate-offset-and-leng
 const UnixFS = require('ipfs-unixfs')
 const errCode = require('err-code')
 
-async function * emitBytes (ipld, node, start, end, streamPosition = 0) {
+async function * emitBytes (ipld, node, start, end, streamPosition = 0, options) {
   // a `raw` node
   if (Buffer.isBuffer(node)) {
     const buf = extractDataFromBlock(node, streamPosition, start, end)
@@ -50,9 +50,9 @@ async function * emitBytes (ipld, node, start, end, streamPosition = 0) {
     if ((start >= childStart && start < childEnd) || // child has offset byte
         (end > childStart && end <= childEnd) || // child has end byte
         (start < childStart && end > childEnd)) { // child is between offset and end bytes
-      const child = await ipld.get(childLink.Hash)
+      const child = await ipld.get(childLink.Hash, options)
 
-      for await (const buf of emitBytes(ipld, child, start, end, streamPosition)) {
+      for await (const buf of emitBytes(ipld, child, start, end, streamPosition, options)) {
         streamPosition += buf.length
 
         yield buf
@@ -76,7 +76,7 @@ const fileContent = (cid, node, unixfs, path, resolve, depth, ipld) => {
     const start = offset
     const end = offset + length
 
-    return emitBytes(ipld, node, start, end)
+    return emitBytes(ipld, node, start, end, 0, options)
   }
 }
 

packages/ipfs-unixfs-exporter/src/resolvers/unixfs-v1/index.js

@@ -19,8 +19,8 @@ const contentExporters = {
   symlink: (cid, node, unixfs, path, resolve, depth, ipld) => {}
 }
 
-const unixFsResolver = async (cid, name, path, toResolve, resolve, depth, ipld) => {
-  const node = await ipld.get(cid)
+const unixFsResolver = async (cid, name, path, toResolve, resolve, depth, ipld, options) => {
+  const node = await ipld.get(cid, options)
   let unixfs
   let next
 
@@ -71,7 +71,7 @@ const unixFsResolver = async (cid, name, path, toResolve, resolve, depth, ipld)
       path,
       cid,
       node,
-      content: contentExporters[unixfs.type](cid, node, unixfs, path, resolve, depth, ipld),
+      content: contentExporters[unixfs.type](cid, node, unixfs, path, resolve, depth, ipld, options),
       unixfs,
       depth
     },

packages/ipfs-unixfs-exporter/src/utils/find-cid-in-shard.js

@@ -62,7 +62,7 @@ const toBucketPath = (position) => {
   return path.reverse()
 }
 
-const findShardCid = async (node, name, ipld, context) => {
+const findShardCid = async (node, name, ipld, context, options) => {
   if (!context) {
     context = {
       rootBucket: new Bucket({
@@ -113,9 +113,9 @@ const findShardCid = async (node, name, ipld, context) => {
 
   context.hamtDepth++
 
-  node = await ipld.get(link.Hash)
+  node = await ipld.get(link.Hash, options)
 
-  return findShardCid(node, name, ipld, context)
+  return findShardCid(node, name, ipld, context, options)
 }
 
 module.exports = findShardCid

packages/ipfs-unixfs-exporter/test/exporter.spec.js

@@ -3,6 +3,7 @@
 
 const chai = require('chai')
 chai.use(require('dirty-chai'))
+chai.use(require('chai-as-promised'))
 const expect = chai.expect
 const IPLD = require('ipld')
 const inMemory = require('ipld-in-memory')
@@ -20,6 +21,7 @@ const all = require('async-iterator-all')
 const last = require('it-last')
 const first = require('async-iterator-first')
 const randomBytes = require('async-iterator-buffer-stream')
+const AbortController = require('abort-controller')
 
 const ONE_MEG = Math.pow(1024, 2)
 
@@ -953,4 +955,35 @@ describe('exporter', () => {
 
     expect(result.toString('utf8')).to.equal('l')
   })
+
+  it('aborts a request', async () => {
+    const abortController = new AbortController()
+
+    // data should not be in IPLD
+    const data = Buffer.from(`hello world '${Math.random()}`)
+    const hash = mh.encode(data, 'sha2-256')
+    const cid = new CID(1, 'dag-pb', hash)
+    const message = `User aborted ${Math.random()}`
+
+    setTimeout(() => {
+      abortController.abort()
+    }, 100)
+
+    // regular test IPLD is offline-only, we need to mimic what happens when
+    // we try to get a block from the network
+    const ipld = {
+      get: (cid, options) => {
+        // promise will never resolve, so reject it when the abort signal is sent
+        return new Promise((resolve, reject) => {
+          options.signal.addEventListener('abort', () => {
+            reject(new Error(message))
+          })
+        })
+      }
+    }
+
+    await expect(exporter(cid, ipld, {
+      signal: abortController.signal
+    })).to.eventually.be.rejectedWith(message)
+  })
 })