fix: blocks may not be in the blockstore yet

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 = 6
 
 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,58 @@ 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()
+    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', 5)
+
+    await engine._processTasks()
+
+    expect(engine).to.have.property('_tasks').that.has.lengthOf(0)
+  })
 })