fix: blocks may not be in the blockstore yet

Because some blockstore implementations batch up write before comitting
them to disk or wherever, they may not be available when we think
they should be.

I noticed this module throwing unhandledPromiseRejections around this
so the change here handles missing blocks by putting the task back
into the queue to process later, with a fairly arbitrary 5x limit on
processing a given task.

Author: achingbrain

package.json

@@ -69,6 +69,7 @@
     "promisify-es6": "^1.0.3",
     "rimraf": "^3.0.0",
     "safe-buffer": "^5.1.2",
+    "sinon": "^9.0.0",
     "stats-lite": "^2.2.0",
     "uuid": "^3.3.2"
   },

src/decision-engine/index.js

@@ -8,6 +8,7 @@ const Ledger = require('./ledger')
 const { logger, groupBy, pullAllWith, uniqWith } = require('../utils')
 
 const MAX_MESSAGE_SIZE = 512 * 1024
+const MAX_TASK_ATTEMPTS = 5
 
 class DecisionEngine {
   constructor (peerId, blockstore, network, stats) {
@@ -80,8 +81,41 @@ class DecisionEngine {
     const cids = entries.map((e) => e.cid)
     const uniqCids = uniqWith((a, b) => a.equals(b), cids)
     const groupedTasks = groupBy(task => task.target.toB58String(), tasks)
+    const unresolvedCids = []
 
-    const blocks = await Promise.all(uniqCids.map(cid => this.blockstore.get(cid)))
+    const blocks = (await Promise.all(
+      uniqCids.map(async cid => {
+        try {
+          const block = await this.blockstore.get(cid)
+
+          return block
+        } catch (err) {
+          this._log.error(`Could not load block for cid ${cid}`, err)
+          unresolvedCids.push(cid)
+        }
+      }))
+    ).filter(Boolean)
+
+    if (blocks.length !== uniqCids.length) {
+      // put any tasks with unresolved CIDs back into the task queue
+      unresolvedCids.forEach(cid => {
+        this._tasks = this._tasks.concat(
+          tasks
+            .filter(task => task.entry.cid.equals(cid))
+            .map(task => {
+              task.attempt = (task.attempt || 0) + 1
+
+              if (task.attempt < MAX_TASK_ATTEMPTS) {
+                return task
+              }
+            })
+            .filter(Boolean)
+        )
+      })
+
+      // run queue again later
+      this._outbox()
+    }
 
     await Promise.all(Object.values(groupedTasks).map(async (tasks) => {
       // all tasks in the group have the same target
@@ -99,8 +133,6 @@ class DecisionEngine {
         this.messageSent(peer, block)
       }
     }))
-
-    this._tasks = []
   }
 
   wantlistForPeer (peerId) {

test/decision-engine/decision-engine.js

@@ -9,9 +9,11 @@ const range = require('lodash.range')
 const difference = require('lodash.difference')
 const flatten = require('lodash.flatten')
 const Block = require('ipfs-block')
+const makeBlock = require('../utils/make-block')
 const CID = require('cids')
 const multihashing = require('multihashing-async')
 const Buffer = require('safe-buffer').Buffer
+const sinon = require('sinon')
 
 const Message = require('../../src/types/message')
 const DecisionEngine = require('../../src/decision-engine')
@@ -197,4 +199,57 @@ describe('Engine', () => {
         .catch(reject)
     })
   })
+
+  it('handles the blockstore not having all the blocks', async () => {
+    const block = await makeBlock()
+
+    const blockstore = {
+      get: sinon.stub().withArgs(block.cid).throws(new Error('Not found'))
+    }
+
+    const engine = new DecisionEngine(await PeerId.create({ bits: 512 }), blockstore)
+    engine._running = true
+
+    engine._tasks.push({
+      entry: {
+        cid: block.cid
+      },
+      target: await PeerId.create({ bits: 512 })
+    })
+
+    await engine._processTasks()
+
+    expect(engine).to.have.nested.deep.property('_tasks[0].entry.cid', block.cid)
+    expect(engine).to.have.nested.deep.property('_tasks[0].attempt', 1)
+  })
+
+  it('does not try to resolve blocks forever', async () => {
+    const block = await makeBlock()
+
+    const blockstore = {
+      get: sinon.stub().withArgs(block.cid).throws(new Error('Not found'))
+    }
+
+    const engine = new DecisionEngine(await PeerId.create({ bits: 512 }), blockstore)
+    engine._running = true
+
+    engine._tasks.push({
+      entry: {
+        cid: block.cid
+      },
+      target: await PeerId.create({ bits: 512 })
+    })
+
+    await engine._processTasks()
+    await engine._processTasks()
+    await engine._processTasks()
+    await engine._processTasks()
+
+    expect(engine).to.have.nested.deep.property('_tasks[0].entry.cid', block.cid)
+    expect(engine).to.have.nested.deep.property('_tasks[0].attempt', 4)
+
+    await engine._processTasks()
+
+    expect(engine).to.have.property('_tasks').that.has.lengthOf(0)
+  })
 })