Add locking for concurrent pin operations

package.json

@@ -72,7 +72,7 @@
     "execa": "^1.0.0",
     "form-data": "^2.3.3",
     "hat": "0.0.3",
-    "interface-ipfs-core": "~0.103.0",
+    "interface-ipfs-core": "ipfs/interface-js-ipfs-core#feat/gc",
     "ipfsd-ctl": "~0.42.0",
     "libp2p-websocket-star": "~0.10.2",
     "ncp": "^2.0.0",
@@ -87,7 +87,7 @@
     "@hapi/joi": "^15.0.1",
     "async": "^2.6.1",
     "async-iterator-all": "0.0.2",
-    "async-iterator-to-pull-stream": "^1.1.0",
+    "async-iterator-to-pull-stream": "^1.3.0",
     "async-iterator-to-stream": "^1.1.0",
     "base32.js": "~0.1.0",
     "bignumber.js": "^8.0.2",
@@ -117,7 +117,7 @@
     "ipfs-bitswap": "~0.24.0",
     "ipfs-block": "~0.8.1",
     "ipfs-block-service": "~0.15.1",
-    "ipfs-http-client": "^32.0.0",
+    "ipfs-http-client": "ipfs/js-ipfs-http-client#feat/gc",
     "ipfs-http-response": "~0.3.0",
     "ipfs-mfs": "~0.11.2",
     "ipfs-multipart": "~0.1.0",
@@ -139,6 +139,7 @@
     "is-pull-stream": "~0.0.0",
     "is-stream": "^2.0.0",
     "iso-url": "~0.4.6",
+    "joi": "^14.3.1",
     "just-flatten-it": "^2.1.0",
     "just-safe-set": "^2.1.0",
     "kind-of": "^6.0.2",
@@ -159,13 +160,15 @@
     "merge-options": "^1.0.1",
     "mime-types": "^2.1.21",
     "mkdirp": "~0.5.1",
+    "mortice": "dirkmc/mortice#fix/read-then-write",
     "multiaddr": "^6.0.5",
     "multiaddr-to-uri": "^4.0.1",
     "multibase": "~0.6.0",
     "multicodec": "~0.5.1",
     "multihashes": "~0.4.14",
     "multihashing-async": "~0.6.0",
     "node-fetch": "^2.3.0",
+    "p-event": "^4.1.0",
     "peer-book": "~0.9.0",
     "peer-id": "~0.12.0",
     "peer-info": "~0.15.0",

src/cli/commands/repo/gc.js

@@ -1,16 +1,37 @@
 'use strict'
 
+const { print } = require('../../utils')
+
 module.exports = {
   command: 'gc',
 
   describe: 'Perform a garbage collection sweep on the repo.',
 
-  builder: {},
+  builder: {
+    quiet: {
+      alias: 'q',
+      desc: 'Write minimal output',
+      type: 'boolean',
+      default: false
+    },
+    'stream-errors': {
+      desc: 'Output individual errors thrown when deleting blocks.',
+      type: 'boolean',
+      default: false
+    }
+  },
 
-  handler (argv) {
-    argv.resolve((async () => {
-      const ipfs = await argv.getIpfs()
-      await ipfs.repo.gc()
+  handler ({ getIpfs, quiet, streamErrors, resolve }) {
+    resolve((async () => {
+      const ipfs = await getIpfs()
+      const res = await ipfs.repo.gc()
+      for (const r of res) {
+        if (r.err) {
+          streamErrors && print(r.err, true, true)
+        } else {
+          print((quiet ? '' : 'Removed ') + r.cid)
+        }
+      }
     })())
   }
 }

src/core/components/block.js

@@ -81,17 +81,19 @@ module.exports = function block (self) {
             cb(null, new Block(block, cid))
           })
         },
-        (block, cb) => self._blockService.put(block, (err) => {
-          if (err) {
-            return cb(err)
-          }
+        (block, cb) => self._gcLock.readLock((_cb) => {
+          self._blockService.put(block, (err) => {
+            if (err) {
+              return _cb(err)
+            }
 
-          if (options.preload !== false) {
-            self._preload(block.cid)
-          }
+            if (options.preload !== false) {
+              self._preload(block.cid)
+            }
 
-          cb(null, block)
-        })
+            _cb(null, block)
+          })
+        }, cb)
       ], callback)
     }),
     rm: promisify((cid, callback) => {
@@ -100,7 +102,8 @@ module.exports = function block (self) {
       } catch (err) {
         return setImmediate(() => callback(errCode(err, 'ERR_INVALID_CID')))
       }
-      self._blockService.delete(cid, callback)
+
+      self._gcLock.writeLock((cb) => self._blockService.delete(cid, cb), callback)
     }),
     stat: promisify((cid, options, callback) => {
       if (typeof options === 'function') {

src/core/components/files-regular/add-pull-stream.js

@@ -112,7 +112,7 @@ function pinFile (file, self, opts, cb) {
   const isRootDir = !file.path.includes('/')
   const shouldPin = pin && isRootDir && !opts.onlyHash && !opts.hashAlg
   if (shouldPin) {
-    return self.pin.add(file.hash, { preload: false }, err => cb(err, file))
+    return self.pin.add(file.hash, { preload: false, lock: false }, err => cb(err, file))
   } else {
     cb(null, file)
   }
@@ -152,7 +152,7 @@ module.exports = function (self) {
     }
 
     opts.progress = progress
-    return pull(
+    return self._gcLock.pullReadLock(() => pull(
       pullMap(content => normalizeContent(content, opts)),
       pullFlatten(),
       pullMap(file => ({
@@ -163,6 +163,6 @@ module.exports = function (self) {
       pullAsyncMap((file, cb) => prepareFile(file, self, opts, cb)),
       pullMap(file => preloadFile(file, self, opts)),
       pullAsyncMap((file, cb) => pinFile(file, self, opts, cb))
-    )
+    ))
   }
 }

src/core/components/object.js

@@ -242,19 +242,21 @@ module.exports = function object (self) {
       }
 
       function next () {
-        self._ipld.put(node, multicodec.DAG_PB, {
-          cidVersion: 0,
-          hashAlg: multicodec.SHA2_256
-        }).then(
-          (cid) => {
-            if (options.preload !== false) {
-              self._preload(cid)
-            }
-
-            callback(null, cid)
-          },
-          (error) => callback(error)
-        )
+        self._gcLock.readLock((cb) => {
+          self._ipld.put(node, multicodec.DAG_PB, {
+            cidVersion: 0,
+            hashAlg: multicodec.SHA2_256
+          }).then(
+            (cid) => {
+              if (options.preload !== false) {
+                self._preload(cid)
+              }
+
+              cb(null, cid)
+            },
+            cb
+          )
+        }, callback)
       }
     }),