fix: Do not load all of a DAG into memory when pinning (#2387)

Port of #2372 into gc branch to ease merging

Author: achingbrain

src/core/components/pin.js

@@ -247,7 +247,7 @@ module.exports = (self) => {
         }
 
         if (type === PinTypes.indirect || type === PinTypes.all) {
-          pinManager.getIndirectKeys((err, indirects) => {
+          pinManager.getIndirectKeys(options, (err, indirects) => {
             if (err) { return callback(err) }
             pins = pins
               // if something is pinned both directly and indirectly,

src/core/components/pin/pin-manager.js

@@ -1,14 +1,14 @@
 /* eslint max-nested-callbacks: ["error", 8] */
 'use strict'
 
-const { DAGNode, DAGLink, util } = require('ipld-dag-pb')
+const { DAGNode, DAGLink } = require('ipld-dag-pb')
 const CID = require('cids')
-const map = require('async/map')
 const series = require('async/series')
 const parallel = require('async/parallel')
 const eachLimit = require('async/eachLimit')
 const waterfall = require('async/waterfall')
 const detectLimit = require('async/detectLimit')
+const queue = require('async/queue')
 const { Key } = require('interface-datastore')
 const errCode = require('err-code')
 const multicodec = require('multicodec')
@@ -43,6 +43,34 @@ class PinManager {
     this.recursivePins = new Set()
   }
 
+  _walkDag ({ cid, preload = false, onCid = () => {} }, cb) {
+    const q = queue(({ cid }, done) => {
+      this.dag.get(cid, { preload }, (err, result) => {
+        if (err) {
+          return done(err)
+        }
+
+        onCid(cid)
+
+        if (result.value.Links) {
+          q.push(result.value.Links.map(link => ({
+            cid: link.Hash
+          })))
+        }
+
+        done()
+      })
+    }, concurrencyLimit)
+    q.drain = () => {
+      cb()
+    }
+    q.error = (err) => {
+      q.kill()
+      cb(err)
+    }
+    q.push({ cid })
+  }
+
   directKeys () {
     return Array.from(this.directPins, key => new CID(key).buffer)
   }
@@ -51,30 +79,21 @@ class PinManager {
     return Array.from(this.recursivePins, key => new CID(key).buffer)
   }
 
-  getIndirectKeys (callback) {
+  getIndirectKeys ({ preload }, callback) {
     const indirectKeys = new Set()
     eachLimit(this.recursiveKeys(), concurrencyLimit, (multihash, cb) => {
-      this.dag._getRecursive(multihash, (err, nodes) => {
-        if (err) {
-          return cb(err)
-        }
-
-        map(nodes, (node, cb) => util.cid(util.serialize(node), {
-          cidVersion: 0
-        }).then(cid => cb(null, cid), cb), (err, cids) => {
-          if (err) {
-            return cb(err)
+      this._walkDag({
+        cid: new CID(multihash),
+        preload: preload || false,
+        onCid: (cid) => {
+          cid = cid.toString()
+
+          // recursive pins pre-empt indirect pins
+          if (!this.recursivePins.has(cid)) {
+            indirectKeys.add(cid)
           }
-
-          cids
-            .map(cid => cid.toString())
-            // recursive pins pre-empt indirect pins
-            .filter(key => !this.recursivePins.has(key))
-            .forEach(key => indirectKeys.add(key))
-
-          cb()
-        })
-      })
+        }
+      }, cb)
     }, (err) => {
       if (err) { return callback(err) }
       callback(null, Array.from(indirectKeys))
@@ -283,6 +302,13 @@ class PinManager {
     })
   }
 
+  fetchCompleteDag (cid, options, callback) {
+    this._walkDag({
+      cid,
+      preload: options.preload
+    }, callback)
+  }
+
   // Returns an error if the pin type is invalid
   static checkPinType (type) {
     if (typeof type !== 'string' || !Object.keys(PinTypes).includes(type)) {