fix: race condition when requesting the same block twice

When we call `blockstore.putMany`, some implementations will batch
up all the `put`s and write them at once.  This means that
`blockstore.has` might not return `true` for a little while - if
another request for a given block comes in before `blockstore.has`
returns `true` it'll get added to the want list.  If the block then
finishes it's batch and finally a remote peer supplies the wanted
block, the notifications that complete the second block request
will never get sent and the process will hang idefinately.

The change made here is to separate the sending of notifications
out from putting things into the blockstore.  If the blockstore has
a block, but the block is still in the wantlist, send notifications
that we now have the block.

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/index.js

@@ -101,6 +101,10 @@ class Bitswap {
     this._updateReceiveCounters(peerId.toB58String(), block, has)
 
     if (has || !wasWanted) {
+      if (wasWanted) {
+        this._sendHaveBlockNotifications(block)
+      }
+
       return
     }
 
@@ -287,16 +291,25 @@ class Bitswap {
 
         yield block
 
-        self.notifications.hasBlock(block)
-        self.engine.receivedBlocks([block.cid])
-        // Note: Don't wait for provide to finish before returning
-        self.network.provide(block.cid).catch((err) => {
-          self._log.error('Failed to provide: %s', err.message)
-        })
+        self._sendHaveBlockNotifications(block)
       }
     }())
   }
 
+  /**
+   * Sends notifications about the arrival of a block
+   *
+   * @param {Block} block
+   */
+  _sendHaveBlockNotifications (block) {
+    this.notifications.hasBlock(block)
+    this.engine.receivedBlocks([block.cid])
+    // Note: Don't wait for provide to finish before returning
+    this.network.provide(block.cid).catch((err) => {
+      this._log.error('Failed to provide: %s', err.message)
+    })
+  }
+
   /**
    * Get the current list of wants.
    *

test/bitswap.js

@@ -5,13 +5,16 @@ const chai = require('chai')
 chai.use(require('dirty-chai'))
 const expect = chai.expect
 const delay = require('delay')
+const PeerId = require('peer-id')
+const sinon = require('sinon')
 
 const Bitswap = require('../src')
 
 const createTempRepo = require('./utils/create-temp-repo-nodejs')
 const createLibp2pNode = require('./utils/create-libp2p-node')
 const makeBlock = require('./utils/make-block')
 const orderedFinish = require('./utils/helpers').orderedFinish
+const Message = require('../src/types/message')
 
 // Creates a repo + libp2pNode + Bitswap with or without DHT
 async function createThing (dht) {
@@ -70,6 +73,54 @@ describe('bitswap without DHT', function () {
 
     finish.assert()
   })
+
+  it('wants a block, receives a block, wants it again before the blockstore has it, receives it after the blockstore has it', async () => {
+    // the block we want
+    const block = await makeBlock()
+
+    // id of a peer with the block we want
+    const peerId = await PeerId.create({ bits: 512 })
+
+    // incoming message with requested block from the other peer
+    const message = new Message(false)
+    message.addEntry(block.cid, 1, false)
+    message.addBlock(block)
+
+    // slow blockstore
+    nodes[0].bitswap.blockstore = {
+      has: sinon.stub().withArgs(block.cid).returns(false),
+      putMany: async function * (source) { // eslint-disable-line require-await
+        yield * source
+      }
+    }
+
+    // add the block to our want list
+    const wantBlockPromise1 = nodes[0].bitswap.get(block.cid)
+
+    // oh look, a peer has sent it to us - this will trigger a `blockstore.putMany` which
+    // for our purposes is a batch operation so `self.blockstore.has(cid)` will still return
+    // false even though we've just yielded a block with that cid
+    await nodes[0].bitswap._receiveMessage(peerId, message)
+
+    // block store did not have it
+    expect(nodes[0].bitswap.blockstore.has.calledWith(block.cid)).to.be.true()
+
+    // another context wants the same block
+    const wantBlockPromise2 = nodes[0].bitswap.get(block.cid)
+
+    // meanwhile the blockstore finishes it's batch
+    nodes[0].bitswap.blockstore.has = sinon.stub().withArgs(block.cid).returns(true)
+
+    // here it comes again
+    await nodes[0].bitswap._receiveMessage(peerId, message)
+
+    // block store had it this time
+    expect(nodes[0].bitswap.blockstore.has.calledWith(block.cid)).to.be.true()
+
+    // both requests should get the block
+    expect(await wantBlockPromise1).to.deep.equal(block)
+    expect(await wantBlockPromise2).to.deep.equal(block)
+  })
 })
 
 describe('bitswap with DHT', function () {