Implement Garbage Collection

package.json

@@ -65,7 +65,7 @@
     "array-shuffle": "^1.0.1",
     "async": "^2.6.1",
     "async-iterator-all": "^1.0.0",
-    "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": "^9.0.0",
@@ -83,6 +83,7 @@
     "debug": "^4.1.0",
     "dlv": "^1.1.3",
     "err-code": "^2.0.0",
+    "explain-error": "^1.0.4",
     "file-type": "^12.0.1",
     "fnv1a": "^1.0.1",
     "fsm-event": "^2.1.0",
@@ -95,7 +96,7 @@
     "ipfs-bitswap": "~0.25.1",
     "ipfs-block": "~0.8.1",
     "ipfs-block-service": "~0.15.2",
-    "ipfs-http-client": "^33.1.0",
+    "ipfs-http-client": "^33.1.1",
     "ipfs-http-response": "~0.3.1",
     "ipfs-mfs": "~0.12.0",
     "ipfs-multipart": "~0.1.1",
@@ -139,6 +140,7 @@
     "merge-options": "^1.0.1",
     "mime-types": "^2.1.21",
     "mkdirp": "~0.5.1",
+    "mortice": "^1.2.2",
     "multiaddr": "^6.1.0",
     "multiaddr-to-uri": "^5.0.0",
     "multibase": "~0.6.0",
@@ -193,6 +195,7 @@
     "ipfsd-ctl": "^0.43.0",
     "libp2p-websocket-star": "~0.10.2",
     "ncp": "^2.0.0",
+    "p-event": "^4.1.0",
     "qs": "^6.5.2",
     "rimraf": "^3.0.0",
     "sinon": "^7.4.0",

src/cli/commands/repo/gc.js

@@ -5,12 +5,31 @@ module.exports = {
 
   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: true
+    }
+  },
 
-  handler (argv) {
-    argv.resolve((async () => {
-      const ipfs = await argv.getIpfs()
-      await ipfs.repo.gc()
+  handler ({ getIpfs, print, 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.message, 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,10 @@ module.exports = function block (self) {
       } catch (err) {
         return setImmediate(() => callback(errCode(err, 'ERR_INVALID_CID')))
       }
-      self._blockService.delete(cid, callback)
+
+      // We need to take a write lock here to ensure that adding and removing
+      // blocks are exclusive operations
+      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

@@ -116,7 +116,9 @@ 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))
+    // Note: addPullStream() has already taken a GC lock, so tell
+    // pin.add() not to take a (second) GC lock
+    return self.pin.add(file.hash, { preload: false, lock: false }, err => cb(err, file))
   } else {
     cb(null, file)
   }
@@ -156,7 +158,7 @@ module.exports = function (self) {
     }
 
     opts.progress = progress
-    return pull(
+    return self._gcLock.pullReadLock(() => pull(
       pullMap(content => normalizeContent(content, opts)),
       pullFlatten(),
       pullMap(file => ({
@@ -167,6 +169,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)
       }
     }),