fix: fix tests and use drain where we drain

Author: achingbrain

package.json

@@ -121,6 +121,7 @@
     "is-ipfs": "~0.6.1",
     "it-all": "^1.0.1",
     "it-concat": "^1.0.0",
+    "it-drain": "^1.0.0",
     "it-first": "^1.0.1",
     "it-glob": "0.0.7",
     "it-last": "^1.0.1",

src/core/components/add/index.js

@@ -4,7 +4,7 @@ const importer = require('ipfs-unixfs-importer')
 const normaliseAddInput = require('ipfs-utils/src/files/normalise-input')
 const { parseChunkerString } = require('./utils')
 const pipe = require('it-pipe')
-const last = require('it-last')
+const drain = require('it-drain')
 
 module.exports = ({ ipld, gcLock, preload, pin, options: constructorOptions }) => {
   const isShardingEnabled = constructorOptions.EXPERIMENTAL && constructorOptions.EXPERIMENTAL.sharding
@@ -112,7 +112,7 @@ function pinFile (pin, opts) {
       if (shouldPin) {
         // Note: addAsyncIterator() has already taken a GC lock, so tell
         // pin.add() not to take a (second) GC lock
-        await last(pin.add(file.cid, {
+        await drain(pin.add(file.cid, {
           preload: false,
           lock: false
         }))

src/core/components/dag/put.js

@@ -2,7 +2,7 @@
 
 const multicodec = require('multicodec')
 const nameToCodec = name => multicodec[name.toUpperCase().replace(/-/g, '_')]
-const last = require('it-last')
+const drain = require('it-drain')
 
 module.exports = ({ ipld, pin, gcLock, preload }) => {
   return async function put (dagNode, options) {
@@ -52,7 +52,7 @@ module.exports = ({ ipld, pin, gcLock, preload }) => {
       })
 
       if (options.pin) {
-        await last(pin.add(cid, {
+        await drain(pin.add(cid, {
           lock: false
         }))
       }

src/core/components/pin/add.js

@@ -14,19 +14,17 @@ module.exports = ({ pinManager, gcLock, dag }) => {
     const pinAdd = async function * () {
       // verify that each hash can be pinned
       for (const cid of cids) {
-        const isPinned = await pinManager.isPinnedWithType(cid, [PinTypes.recursive, PinTypes.direct])
-        const pinned = isPinned.pinned
+        const { reason } = await pinManager.isPinnedWithType(cid, [PinTypes.recursive, PinTypes.direct])
 
-        if (pinned) {
-          throw new Error(`${cid} already pinned with type ${isPinned.reason}`)
+        if (reason === 'recursive' && !recursive) {
+          // only disallow trying to override recursive pins
+          throw new Error(`${cid} already pinned recursively`)
         }
 
-        if (!pinned) {
-          if (recursive) {
-            await pinManager.pinRecursively(cid)
-          } else {
-            await pinManager.pinDirectly(cid)
-          }
+        if (recursive) {
+          await pinManager.pinRecursively(cid)
+        } else {
+          await pinManager.pinDirectly(cid)
         }
 
         yield { cid }

src/http/api/resources/dag.js

@@ -14,7 +14,7 @@ const {
 const all = require('it-all')
 const log = debug('ipfs:http-api:dag')
 log.error = debug('ipfs:http-api:dag:error')
-const last = require('it-last')
+const drain = require('it-drain')
 
 const IpldFormats = {
   get [multicodec.RAW] () {
@@ -253,7 +253,7 @@ exports.put = {
     }
 
     if (request.query.pin) {
-      await last(ipfs.pin.add(cid))
+      await drain(ipfs.pin.add(cid))
     }
 
     return h.response({

test/core/gc.spec.js

@@ -4,6 +4,7 @@
 
 const { expect } = require('interface-ipfs-core/src/utils/mocha')
 const last = require('it-last')
+const drain = require('it-drain')
 const factory = require('../utils/factory')
 const pEvent = require('p-event')
 
@@ -230,8 +231,8 @@ describe.skip('gc', function () {
       const cid2 = (await ipfs.block.put(Buffer.from('block to pin rm 2'), null)).cid
 
       // Pin blocks
-      await last(ipfs.pin.add(cid1))
-      await last(ipfs.pin.add(cid2))
+      await drain(ipfs.pin.add(cid1))
+      await drain(ipfs.pin.add(cid2))
 
       // Unpin first block
       // Note: pin rm will take a read lock

test/http-api/inject/pin.js

@@ -52,7 +52,14 @@ module.exports = (http) => {
       })
 
       it('unpins recursive pins', async () => {
-        const res = await api.inject({
+        let res = await api.inject({
+          method: 'POST',
+          url: `/api/v0/pin/add?arg=${pins.root1}`
+        })
+
+        expect(res.statusCode).to.equal(200)
+
+        res = await api.inject({
           method: 'POST',
           url: `/api/v0/pin/rm?arg=${pins.root1}`
         })
@@ -71,7 +78,7 @@ module.exports = (http) => {
 
         res = await api.inject({
           method: 'POST',
-          url: `/api/v0/pin/rm?arg=${pins.root1}&recursive=false`
+          url: `/api/v0/pin/rm?arg=${pins.root1}`
         })
 
         expect(res.statusCode).to.equal(200)