chore: upgrade repo to streaming api (#3041)

The latest interface-datastore contains added streaming apis which will let us use fewer resources when importing large datasets.

This PR upgrades ipfs-repo and all datastore-* deps. Also bitswap.

Author: achingbrain

examples/custom-ipfs-repo/package.json

@@ -9,9 +9,9 @@
   },
   "license": "MIT",
   "dependencies": {
-    "datastore-fs": "^0.9.1",
+    "datastore-fs": "^1.1.0",
     "ipfs": "^0.44.0",
-    "ipfs-repo": "^2.0.1",
+    "ipfs-repo": "^3.0.0",
     "it-all": "^1.0.1"
   },
   "devDependencies": {

packages/interface-ipfs-core/package.json

@@ -31,6 +31,7 @@
   ],
   "license": "MIT",
   "dependencies": {
+    "abort-controller": "^3.0.0",
     "buffer": "^5.6.0",
     "chai": "^4.2.0",
     "chai-as-promised": "^7.1.1",

packages/interface-ipfs-core/src/bitswap/utils.js

@@ -21,4 +21,26 @@ async function waitForWantlistKey (ipfs, key, opts = {}) {
   throw new Error(`Timed out waiting for ${key} in wantlist`)
 }
 
+async function waitForWantlistKeyToBeRemoved (ipfs, key, opts = {}) {
+  opts.timeout = opts.timeout || 10000
+  opts.interval = opts.interval || 100
+
+  const end = Date.now() + opts.timeout
+
+  while (Date.now() < end) {
+    const list = await ipfs.bitswap.wantlist(opts.peerId)
+
+    if (list.some(cid => cid.toString() === key)) {
+      await delay(opts.interval)
+
+      continue
+    }
+
+    return
+  }
+
+  throw new Error(`Timed out waiting for ${key} to be removed from wantlist`)
+}
+
 module.exports.waitForWantlistKey = waitForWantlistKey
+module.exports.waitForWantlistKeyToBeRemoved = waitForWantlistKeyToBeRemoved

packages/interface-ipfs-core/src/bitswap/wantlist.js

@@ -2,9 +2,12 @@
 'use strict'
 
 const { getDescribe, getIt, expect } = require('../utils/mocha')
-const { waitForWantlistKey } = require('./utils')
+const { waitForWantlistKey, waitForWantlistKeyToBeRemoved } = require('./utils')
 const { isWebWorker } = require('ipfs-utils/src/env')
 const testTimeout = require('../utils/test-timeout')
+const AbortController = require('abort-controller')
+const CID = require('cids')
+const delay = require('delay')
 
 /** @typedef { import("ipfsd-ctl/src/factory") } Factory */
 /**
@@ -57,5 +60,52 @@ module.exports = (common, options) => {
 
       return expect(node.api.bitswap.stat()).to.eventually.be.rejected()
     })
+
+    it('should remove blocks from the wantlist when requests are cancelled', async () => {
+      const controller = new AbortController()
+      const cid = new CID('QmSoLPppuBtQSGwKDZT2M73ULpjvfd3aZ6ha4oFGL1KaGa')
+
+      const getPromise = ipfsA.dag.get(cid, {
+        signal: controller.signal
+      })
+
+      await waitForWantlistKey(ipfsA, cid.toString())
+
+      controller.abort()
+
+      await expect(getPromise).to.eventually.be.rejectedWith(/aborted/)
+
+      await waitForWantlistKeyToBeRemoved(ipfsA, cid.toString())
+    })
+
+    it('should keep blocks in the wantlist when only one request is cancelled', async () => {
+      const controller = new AbortController()
+      const otherController = new AbortController()
+      const cid = new CID('QmSoLPppuBtQSGwKDZT2M73ULpjvfd3aZ6ha4oFGL1Kaaa')
+
+      const getPromise = ipfsA.dag.get(cid, {
+        signal: controller.signal
+      })
+      const otherGetPromise = ipfsA.dag.get(cid, {
+        signal: otherController.signal
+      })
+
+      await waitForWantlistKey(ipfsA, cid.toString())
+
+      controller.abort()
+
+      await expect(getPromise).to.eventually.be.rejectedWith(/aborted/)
+
+      await delay(1000)
+
+      // cid should still be in the wantlist
+      await waitForWantlistKey(ipfsA, cid.toString())
+
+      otherController.abort()
+
+      await expect(otherGetPromise).to.eventually.be.rejectedWith(/aborted/)
+
+      await waitForWantlistKeyToBeRemoved(ipfsA, cid.toString())
+    })
   })
 }

packages/ipfs/.aegir.js

@@ -15,7 +15,7 @@ let sigServerB
 let ipfsdServer
 
 module.exports = {
-  bundlesize: { maxSize: '601kB' },
+  bundlesize: { maxSize: '446kB' },
   karma: {
     files: [{
       pattern: 'node_modules/interface-ipfs-core/test/fixtures/**/*',

packages/ipfs/package.json

@@ -80,9 +80,9 @@
     "cids": "^0.8.0",
     "class-is": "^1.1.0",
     "dag-cbor-links": "^1.3.3",
-    "datastore-core": "^1.0.0",
-    "datastore-level": "^1.0.0",
-    "datastore-pubsub": "^0.3.1",
+    "datastore-core": "^1.1.0",
+    "datastore-level": "^1.1.0",
+    "datastore-pubsub": "^0.3.2",
     "debug": "^4.1.0",
     "dlv": "^1.1.3",
     "err-code": "^2.0.0",
@@ -92,13 +92,13 @@
     "hamt-sharding": "^1.0.0",
     "hapi-pino": "^6.1.0",
     "hashlru": "^2.3.0",
-    "interface-datastore": "^0.8.3",
-    "ipfs-bitswap": "^0.27.2",
-    "ipfs-block-service": "^0.16.0",
+    "interface-datastore": "^1.0.2",
+    "ipfs-bitswap": "^1.0.0",
+    "ipfs-block-service": "^0.17.1",
     "ipfs-core-utils": "^0.2.3",
     "ipfs-http-client": "^44.1.0",
     "ipfs-http-response": "^0.5.0",
-    "ipfs-repo": "^2.0.1",
+    "ipfs-repo": "^3.0.0",
     "ipfs-unixfs": "^1.0.2",
     "ipfs-unixfs-exporter": "^2.0.1",
     "ipfs-unixfs-importer": "^2.0.1",

packages/ipfs/src/core/components/block/get.js

@@ -12,6 +12,6 @@ module.exports = ({ blockService, preload }) => {
       preload(cid)
     }
 
-    return blockService.get(cid)
+    return blockService.get(cid, options)
   })
 }

packages/ipfs/src/core/components/dag/get.js

@@ -12,7 +12,7 @@ module.exports = ({ ipld, preload }) => {
     }
 
     if (path == null || path === '/') {
-      const value = await ipld.get(cid)
+      const value = await ipld.get(cid, options)
 
       return {
         value,

packages/ipfs/src/core/components/dag/put.js

@@ -48,7 +48,8 @@ module.exports = ({ ipld, pin, gcLock, preload }) => {
     try {
       const cid = await ipld.put(dagNode, options.format, {
         hashAlg: options.hashAlg,
-        cidVersion: options.version
+        cidVersion: options.version,
+        signal: options.signal
       })
 
       if (options.pin) {