Add ifAvailable flag to core.download() (#110)

* Add `ifAvailable` flag to `core.download()`

* Add initial tests

* Fix assertions

* Check `ifAvailable` ranges in `updatePeer()` and `updateAll()`

* if async work is happening, defer ifavailable decision

* test a -> b -> c replication with ifavail

* more tests

* fix no solo

Co-authored-by: Mathias Buus <[email protected]>

Author: kasperisager

lib/replicator.js

@@ -102,12 +102,13 @@ class BlockRequest extends Attachable {
 }
 
 class RangeRequest extends Attachable {
-  constructor (ranges, start, end, linear, blocks) {
+  constructor (ranges, start, end, linear, ifAvailable, blocks) {
     super()
 
     this.start = start
     this.end = end
     this.linear = linear
+    this.ifAvailable = ifAvailable
     this.blocks = blocks
     this.ranges = ranges
 
@@ -172,6 +173,10 @@ class InflightTracker {
     this._free = []
   }
 
+  get idle () {
+    return this._requests.length === this._free.length
+  }
+
   * [Symbol.iterator] () {
     for (const req of this._requests) {
       if (req !== null) yield req
@@ -260,6 +265,7 @@ class Peer {
 
     this.inflight = 0
     this.maxInflight = DEFAULT_MAX_INFLIGHT
+    this.dataProcessing = 0
 
     this.canUpgrade = true
 
@@ -515,6 +521,8 @@ class Peer {
 
     if (reorg === true) return this.replicator._onreorgdata(this, req, data)
 
+    this.dataProcessing++
+
     try {
       if (!matchingRequest(req, data) || !(await this.core.verify(data, this))) {
         this.replicator._onnodata(this, req)
@@ -523,6 +531,8 @@ class Peer {
     } catch (err) {
       this.replicator._onnodata(this, req)
       throw err
+    } finally {
+      this.dataProcessing--
     }
 
     this.replicator._ondata(this, req, data)
@@ -909,13 +919,21 @@ module.exports = class Replicator {
     return ref
   }
 
-  addRange (session, { start = 0, end = -1, length = toLength(start, end), blocks = null, linear = false } = {}) {
+  addRange (session, { start = 0, end = -1, length = toLength(start, end), blocks = null, linear = false, ifAvailable = false } = {}) {
     if (blocks !== null) { // if using blocks, start, end just acts as frames around the blocks array
       start = 0
       end = length = blocks.length
     }
 
-    const r = new RangeRequest(this._ranges, start, length === -1 ? -1 : start + length, linear, blocks)
+    const r = new RangeRequest(
+      this._ranges,
+      start,
+      length === -1 ? -1 : start + length,
+      linear,
+      ifAvailable,
+      blocks
+    )
+
     const ref = r.attach(session)
 
     this._ranges.push(r)
@@ -1115,6 +1133,14 @@ module.exports = class Replicator {
     return true
   }
 
+  _resolveRangeRequest (req, index) {
+    const head = this._ranges.pop()
+
+    if (index < this._ranges.length) this._ranges[index] = head
+
+    req.resolve(true)
+  }
+
   _clearInflightBlock (tracker, req) {
     const isBlock = tracker === this._blocks
     const index = isBlock === true ? req.block.index : req.hash.index / 2
@@ -1168,14 +1194,9 @@ module.exports = class Replicator {
         while (r.start < r.end && this.core.bitfield.get(mapIndex(r.blocks, r.start)) === true) r.start++
         while (r.start < r.end && this.core.bitfield.get(mapIndex(r.blocks, r.end - 1)) === true) r.end--
 
-        if (r.end === -1 || r.start < r.end) continue
-
-        if (i < this._ranges.length - 1) this._ranges[i] = this._ranges.pop()
-        else this._ranges.pop()
-
-        i--
-
-        r.resolve(true)
+        if (r.end !== -1 && r.start >= r.end) {
+          this._resolveRangeRequest(r, i--)
+        }
       }
 
       for (let i = 0; i < this._seeks.length; i++) {
@@ -1201,7 +1222,7 @@ module.exports = class Replicator {
         else s.resolve(res)
       }
 
-      this.updateAll()
+      if (this._inflight.idle) this.updateAll()
 
       // No additional updates scheduled - return
       if (--this._updatesPending === 0) return
@@ -1210,6 +1231,22 @@ module.exports = class Replicator {
     }
   }
 
+  _maybeResolveIfAvailableRanges () {
+    if (this._ifAvailable > 0 || !this._inflight.idle || !this._ranges.length) return
+
+    for (let i = 0; i < this.peers.length; i++) {
+      if (this.peers[i].dataProcessing > 0) return
+    }
+
+    for (let i = 0; i < this._ranges.length; i++) {
+      const r = this._ranges[i]
+
+      if (r.ifAvailable) {
+        this._resolveRangeRequest(r, i--)
+      }
+    }
+  }
+
   _clearRequest (peer, req) {
     if (req.block !== null) {
       this._clearInflightBlock(this._blocks, req)
@@ -1407,6 +1444,7 @@ module.exports = class Replicator {
     while (this._updatePeerNonPrimary(peer) === true);
 
     this._checkUpgradeIfAvailable()
+    this._maybeResolveIfAvailableRanges()
   }
 
   updateAll () {
@@ -1434,6 +1472,7 @@ module.exports = class Replicator {
     }
 
     this._checkUpgradeIfAvailable()
+    this._maybeResolveIfAvailableRanges()
   }
 
   attachTo (protomux) {

test/replicate.js

@@ -638,3 +638,91 @@ test('non-sparse replication', async function (t) {
 
   t.is(contiguousLength, b.length)
 })
+
+test('download blocks if available', async function (t) {
+  const a = await create()
+  const b = await create(a.key)
+
+  replicate(a, b, t)
+
+  await a.append(['a', 'b', 'c', 'd', 'e'])
+
+  let d = 0
+  b.on('download', () => d++)
+
+  const r = b.download({ blocks: [1, 3, 6], ifAvailable: true })
+  await r.downloaded()
+
+  t.is(d, 2)
+})
+
+test('download range if available', async function (t) {
+  const a = await create()
+  const b = await create(a.key)
+
+  replicate(a, b, t)
+
+  await a.append(['a', 'b', 'c', 'd', 'e'])
+
+  let d = 0
+  b.on('download', () => d++)
+
+  const r = b.download({ start: 2, end: 6, ifAvailable: true })
+  await r.downloaded()
+
+  t.is(d, 3)
+})
+
+test('download blocks if available, destroy midway', async function (t) {
+  const a = await create()
+  const b = await create(a.key)
+
+  const s = replicate(a, b, t)
+
+  await a.append(['a', 'b', 'c', 'd', 'e'])
+
+  let d = 0
+  b.on('download', () => {
+    if (d++ === 0) unreplicate(s)
+  })
+
+  const r = b.download({ blocks: [1, 3, 6], ifAvailable: true })
+  await r.downloaded()
+
+  t.pass('range resolved')
+})
+
+test('download blocks available from when only a partial set is available', async function (t) {
+  const a = await create()
+  const b = await create(a.key)
+  const c = await create(a.key)
+
+  replicate(a, b, t)
+  replicate(b, c, t)
+
+  await a.append(['a', 'b', 'c', 'd', 'e'])
+
+  await b.get(2)
+  await b.get(3)
+
+  const r = c.download({ start: 0, end: -1, ifAvailable: true })
+  await r.downloaded()
+
+  t.ok(!(await c.has(0)))
+  t.ok(!(await c.has(1)))
+  t.ok(await c.has(2))
+  t.ok(await c.has(3))
+  t.ok(!(await c.has(4)))
+})
+
+test('download range resolves immediately if no peers', async function (t) {
+  const a = await create()
+  const b = await create(a.key)
+
+  // no replication
+
+  const r = b.download({ start: 0, end: 5, ifAvailable: true })
+  await r.downloaded()
+
+  t.pass('range resolved')
+})