Add peer.remoteContiguousLength

lib/replicator.js

@@ -325,6 +325,10 @@ class Peer {
  replicator._ifAvailable++
  }
 
+ get remoteContiguousLength () {
+ return this.remoteBitfield.findFirst(false, 0)
+ }
+
  getMaxInflight () {
  const stream = this.stream.rawStream
  if (!stream.udx) return Math.min(this.inflightRange[1], this.inflightRange[0] * 3)
@@ -680,8 +684,12 @@ class Peer {
 
  onrange ({ drop, start, length }) {
  const has = drop === false
+ const rangeStart = this.remoteBitfield.findFirst(!has, start)
+ const rangeLength = length - (rangeStart - start)
 
- this.remoteBitfield.setRange(start, length, has)
+ if (rangeLength > 0) {
+ this.remoteBitfield.setRange(rangeStart, rangeLength, has)
+ }
 
  if (drop === false) this._update()
  }
@@ -912,6 +920,8 @@ class Peer {
  }
 
  _maybeWant (start, length = 1) {
+ if (start + length <= this.remoteContiguousLength) return
+
  let i = Math.floor(start / DEFAULT_SEGMENT_SIZE)
  const n = Math.ceil((start + length) / DEFAULT_SEGMENT_SIZE)
 

test/remote-length.js

@@ -0,0 +1,68 @@
+const test = require('brittle')
+const { create, replicate } = require('./helpers')
+
+test('basic peer remote contiguous length', async function (t) {
+ t.plan(2)
+
+ const a = await create()
+ const b = await create(a.key)
+
+ replicate(a, b, t)
+
+ await a.append('a')
+ await b.get(0)
+ await new Promise(resolve => setTimeout(resolve, 100))
+ t.is(getPeer(a, b).remoteContiguousLength, 1)
+
+ await a.append('b')
+ await b.get(1)
+ await new Promise(resolve => setTimeout(resolve, 100))
+ t.is(getPeer(a, b).remoteContiguousLength, 2)
+})
+
+test('peer remote contiguous length', async function (t) {
+ t.plan(3)
+
+ const a = await create()
+ const b = await create(a.key)
+
+ replicate(a, b, t)
+
+ await a.append('a')
+ await b.get(0)
+ await new Promise(resolve => setTimeout(resolve, 100))
+ t.is(getPeer(a, b).remoteContiguousLength, 1)
+
+ await a.append(['d', 'e'])
+ await b.get(2)
+ await new Promise(resolve => setTimeout(resolve, 100))
+ t.is(getPeer(a, b).remoteContiguousLength, 1)
+
+ await b.get(1)
+ await new Promise(resolve => setTimeout(resolve, 100))
+ t.is(getPeer(a, b).remoteContiguousLength, 3)
+})
+
+test('peer truncates the remote contiguous length', async function (t) {
+ t.plan(2)
+
+ const a = await create()
+ const b = await create(a.key)
+
+ replicate(a, b, t)
+
+ await a.append(['a', 'b'])
+ await b.get(0)
+ await b.get(1)
+ await new Promise(resolve => setTimeout(resolve, 100))
+ t.is(getPeer(a, b).remoteContiguousLength, 2)
+
+ await a.truncate(1)
+ await new Promise(resolve => setTimeout(resolve, 100))
+ t.is(getPeer(a, b).remoteContiguousLength, 1)
+})
+
+// "A" wants to know if "B" is finished syncing, so find the corresponding peer
+function getPeer (a, b) {
+ return a.replicator.peers.find(peer => peer.remotePublicKey.equals(b.replicator.peers[0].stream.publicKey))
+}