Merge pull request #85 from ipfs/lock

fix(blockstore: lock getStream to avoid race issues

Author: daviddias

package.json

@@ -47,8 +47,8 @@
     "ipfs-block": "^0.3.0",
     "lock": "^0.1.3",
     "multihashes": "^0.2.2",
+    "pull-defer": "^0.2.2",
     "pull-stream": "^3.4.5",
-    "pull-through": "^1.0.18",
     "pull-write": "^1.1.0",
     "run-parallel": "^1.1.6",
     "run-series": "^1.1.4",
@@ -66,4 +66,4 @@
     "greenkeeperio-bot <[email protected]>",
     "nginnever <[email protected]>"
   ]
-}
+}

src/stores/blockstore.js

@@ -7,7 +7,7 @@ const base32 = require('base32.js')
 const path = require('path')
 const write = require('pull-write')
 const parallel = require('run-parallel')
-const through = require('pull-through')
+const defer = require('pull-defer/source')
 
 const PREFIX_LENGTH = 5
 
@@ -52,19 +52,25 @@ exports.setUp = (basePath, BlobStore, locks) => {
       }
 
       const p = multihashToPath(key, extension)
+      const deferred = defer()
+
+      lock(p, (release) => {
+        const ext = extension === 'data' ? 'protobuf' : extension
+        pull(
+          store.read(p),
+          pull.collect(release((err, data) => {
+            if (err) {
+              return deferred.abort(err)
+            }
+
+            deferred.resolve(pull.values([
+              new Block(Buffer.concat(data), ext)
+            ]))
+          }))
+        )
+      })
 
-      const ext = extension === 'data' ? 'protobuf' : extension
-      let data = []
-
-      return pull(
-        store.read(p),
-        through(function (values) {
-          data = data.concat(values)
-        }, function () {
-          this.queue(new Block(Buffer.concat(data), ext))
-          this.queue(null)
-        })
-      )
+      return deferred
     },
 
     putStream () {
@@ -75,7 +81,10 @@ exports.setUp = (basePath, BlobStore, locks) => {
       const sink = write((blocks, cb) => {
         parallel(blocks.map((block) => (cb) => {
           writeBlock(block, (err, meta) => {
-            if (err) return cb(err)
+            if (err) {
+              return cb(err)
+            }
+
             if (push) {
               const read = push
               push = null
@@ -94,7 +103,9 @@ exports.setUp = (basePath, BlobStore, locks) => {
 
       const source = (end, cb) => {
         if (end) ended = end
-        if (ended) return cb(ended)
+        if (ended) {
+          return cb(ended)
+        }
 
         if (written.length) {
           return cb(null, written.shift())

test/blockstore-test.js

@@ -109,7 +109,7 @@ module.exports = (repo) => {
 
       it('massive read', (done) => {
         parallel(_.range(20 * 100).map((i) => (cb) => {
-          const j = i % 20
+          const j = i % blockCollection.length
           pull(
             repo.blockstore.getStream(blockCollection[j].key),
             pull.collect((err, meta) => {