fix: survive bad network requests (#222)

* fix: survive not being able to send messages to peers

Builds on the work in #221

I see intermittant but frequent CI errors with js-ipfs, usually after
a test has finished and the ndoes are being torn down.  Getting it
to do some additional logging reveals bitswap is crashing when it
cannot send a message to a remote peer due to the libp2p dial failing:

```console
ipfs: [stdout] Error: stream ended before 1 bytes became available
ipfs:     at /home/travis/build/ipfs/js-ipfs/node_modules/it-reader/index.js:37:9
ipfs:     at processTicksAndRejections (internal/process/task_queues.js:97:5)
ipfs:     at async /home/travis/build/ipfs/js-ipfs/node_modules/it-length-prefixed/src/decode.js:80:20
ipfs:     at async oneChunk (/home/travis/build/ipfs/js-ipfs/node_modules/multistream-select/src/multistream.js:12:20)
ipfs:     at async Object.exports.read (/home/travis/build/ipfs/js-ipfs/node_modules/multistream-select/src/multistream.js:34:15)
ipfs:     at async module.exports (/home/travis/build/ipfs/js-ipfs/node_modules/multistream-select/src/select.js:21:19)
ipfs:     at async ClassIsWrapper.newStream [as _newStream] (/home/travis/build/ipfs/js-ipfs/node_modules/libp2p/src/upgrader.js:251:40)
ipfs:     at async ClassIsWrapper.newStream (/home/travis/build/ipfs/js-ipfs/node_modules/libp2p-interfaces/src/connection/connection.js:172:34)
ipfs:     at async Network.sendMessage (/home/travis/build/ipfs/js-ipfs/node_modules/ipfs-bitswap/src/network.js:147:34)
ipfs:     at async DecisionEngine._processTasks (/home/travis/build/ipfs/js-ipfs/node_modules/ipfs-bitswap/src/decision-engine/index.js:124:5) {
ipfs:   code: 'ERR_UNSUPPORTED_PROTOCOL',
ipfs:   buffer: BufferList { _bufs: [], length: 0 }
ipfs: }
```

This PR adds a try/catch around network send operations.  At the moment
it just dumps the request, I'm not sure if we want to add a retry in
there or something.

Author: achingbrain

src/decision-engine/index.js

@@ -120,17 +120,21 @@ class DecisionEngine {
       return
     }
 
-    // Send the message
-    await this.network.sendMessage(peerId, msg)
+    try {
+      // Send the message
+      await this.network.sendMessage(peerId, msg)
+
+      // Peform sent message accounting
+      for (const block of blocks.values()) {
+        this.messageSent(peerId, block)
+      }
+    } catch (err) {
+      this._log.error(err)
+    }
 
     // Free the tasks up from the request queue
     this._requestQueue.tasksDone(peerId, tasks)
 
-    // Peform sent message accounting
-    for (const block of blocks.values()) {
-      this.messageSent(peerId, block)
-    }
-
     // Trigger the next round of task processing
     this._scheduleProcessTasks()
   }

test/decision-engine/decision-engine.js

@@ -683,4 +683,50 @@ describe('Engine', () => {
       }
     }
   })
+
+  it('survives not being able to send a message to peer', async () => {
+    let r
+    const failToSendPromise = new Promise((resolve) => {
+      r = resolve
+    })
+
+    const network = mockNetwork()
+    network.sendMessage = () => {
+      r()
+      throw new Error('Something is b0rken')
+    }
+
+    // who is in the network
+    const us = await newEngine(network)
+    const them = await newEngine()
+
+    // add a block to our blockstore
+    const data = Buffer.from(`this is message ${Date.now()}`)
+    const hash = await multihashing(data, 'sha2-256')
+    const cid = new CID(hash)
+    const block = new Block(data, cid)
+    await us.engine.blockstore.put(block)
+
+    // receive a message with a want for our block
+    await us.engine.messageReceived(them.peer, {
+      blocks: [],
+      wantlist: [{
+        cid,
+        priority: 1,
+        wantType: 'wanty'
+      }]
+    })
+
+    // should have added a task for the remote peer
+    const tasks = us.engine._requestQueue._byPeer.get(them.peer.toB58String())
+
+    expect(tasks).to.have.property('_pending').that.has.property('length', 1)
+
+    // wait for us.network.sendMessage to be called
+    await failToSendPromise
+
+    // should be done processing
+    expect(tasks).to.have.property('_pending').that.has.property('length', 0)
+    expect(tasks).to.have.property('_active').that.has.property('size', 0)
+  })
 })