Broadcast [0, contig) range on replicator open and add want messages (#114)

* Broadcast `[0, contig)` range on replicator open

* Add failing test case

* Test ranges as well

* First draft of "want" flow

* Test discrete ranges

* Handle discrete ranges

* Reduce test duplication

* Test seeks

* GC remaining wants when peers are done processing

* Only broadcast `contig` if `> 0`

* Fix tests

* Simplify wants

* Send bitfield pages on want

* Increase default segment size

* Broadcast non-sparse length to peers

* Set flag

* Brittle workaround

Author: kasperisager

index.js

@@ -527,6 +527,18 @@ module.exports = class Hypercore extends EventEmitter {
           s.emit('append')
         }
       }
+
+      const contig = this.core.header.contiguousLength
+
+      // When the contig length catches up, broadcast the non-sparse length to peers
+      if (appendedNonSparse && contig === this.core.tree.length) {
+        for (const peer of this.peers) {
+          if (peer.broadcastedNonSparse) continue
+
+          peer.broadcastRange(0, contig)
+          peer.broadcastedNonSparse = true
+        }
+      }
     }
 
     if (bitfield) {
@@ -662,29 +674,25 @@ module.exports = class Hypercore extends EventEmitter {
   }
 
   async _get (index, opts) {
-    let req
+    let block
 
     if (this.core.bitfield.get(index)) {
-      req = this.core.blocks.get(index)
+      block = this.core.blocks.get(index)
 
-      if (this.cache) this.cache.set(index, req)
+      if (this.cache) this.cache.set(index, block)
     } else {
       if (opts && opts.wait === false) return null
       if (opts && opts.onwait) opts.onwait(index, this)
       else if (this.onwait) this.onwait(index, this)
 
       const activeRequests = (opts && opts.activeRequests) || this.activeRequests
 
-      req = this._cacheOnResolve(
-        index,
-        this.replicator
-          .addBlock(activeRequests, index)
-          .promise,
-        this.core.tree.fork
-      )
+      const req = this.replicator.addBlock(activeRequests, index)
+
+      block = this._cacheOnResolve(index, req.promise, this.core.tree.fork)
     }
 
-    return req
+    return block
   }
 
   async _cacheOnResolve (index, req, fork) {
@@ -725,7 +733,9 @@ module.exports = class Hypercore extends EventEmitter {
 
   async _download (range) {
     if (this.opened === false) await this.opening
+
     const activeRequests = (range && range.activeRequests) || this.activeRequests
+
     return this.replicator.addRange(activeRequests, range)
   }
 

lib/bitfield.js

@@ -98,12 +98,6 @@ module.exports = class Bitfield {
     }
   }
 
-  // Should prob be removed, when/if we re-add compression
-  page (i) {
-    const p = this.pages.get(i)
-    return p ? p.bitfield : new Uint32Array(1024)
-  }
-
   clear () {
     return new Promise((resolve, reject) => {
       this.storage.del(0, Infinity, (err) => {

lib/replicator.js

@@ -7,6 +7,7 @@ const m = require('./messages')
 const caps = require('./caps')
 
 const DEFAULT_MAX_INFLIGHT = 32
+const DEFAULT_SEGMENT_SIZE = 128 * 1024 // 128 KiB
 
 class Attachable {
   constructor () {
@@ -286,6 +287,9 @@ class Peer {
     this.remoteDownloading = true
     this.remoteSynced = false
 
+    this.segmentsWanted = new Set()
+    this.broadcastedNonSparse = false
+
     this.lengthAcked = 0
 
     this.extensions = new Map()
@@ -356,13 +360,13 @@ class Peer {
 
     this.sendSync()
 
-    const p = pages(this.core)
+    const contig = this.core.header.contiguousLength
+    if (contig > 0) {
+      this.broadcastRange(0, contig, false)
 
-    for (let index = 0; index < p.length; index++) {
-      this.wireBitfield.send({
-        start: index * this.core.bitfield.pageSize,
-        bitfield: p[index]
-      })
+      if (contig === this.core.tree.length) {
+        this.broadcastedNonSparse = true
+      }
     }
 
     this.replicator._ifAvailable--
@@ -552,8 +556,8 @@ class Peer {
     this.replicator._onnodata(this, req)
   }
 
-  onwant () {
-    // TODO
+  onwant ({ start, length }) {
+    this.replicator._onwant(this, start, length)
   }
 
   onunwant () {
@@ -672,6 +676,7 @@ class Peer {
       return true
     }
 
+    this._maybeWant(s.seeker.start, len)
     return false
   }
 
@@ -683,7 +688,10 @@ class Peer {
   _requestBlock (b) {
     const { length, fork } = this.core.tree
 
-    if (this.remoteBitfield.get(b.index) === false || fork !== this.remoteFork) return false
+    if (this.remoteBitfield.get(b.index) === false || fork !== this.remoteFork) {
+      this._maybeWant(b.index)
+      return false
+    }
 
     const req = this._makeRequest(b.index >= length)
     if (req === null) return false
@@ -741,6 +749,7 @@ class Peer {
       return true
     }
 
+    this._maybeWant(r.start, len)
     return false
   }
 
@@ -778,9 +787,25 @@ class Peer {
       return true
     }
 
+    this._maybeWant(f.batch.want.start, len)
     return false
   }
 
+  _maybeWant (start, length = 1) {
+    let i = Math.floor(start / DEFAULT_SEGMENT_SIZE)
+    const n = Math.ceil((start + length) / DEFAULT_SEGMENT_SIZE)
+
+    for (; i < n; i++) {
+      if (this.segmentsWanted.has(i)) continue
+      this.segmentsWanted.add(i)
+
+      this.wireWant.send({
+        start: i * DEFAULT_SEGMENT_SIZE,
+        length: DEFAULT_SEGMENT_SIZE
+      })
+    }
+  }
+
   async _send (req) {
     const fork = this.core.tree.fork
 
@@ -1297,6 +1322,28 @@ module.exports = class Replicator {
     else this.updatePeer(peer)
   }
 
+  _onwant (peer, start, length) {
+    length = Math.min(length, this.core.tree.length - start)
+
+    peer.protomux.cork()
+
+    let i = Math.floor(start / this.core.bitfield.pageSize)
+    const n = Math.ceil((start + length) / this.core.bitfield.pageSize)
+
+    for (; i < n; i++) {
+      const p = this.core.bitfield.pages.get(i)
+
+      if (p) {
+        peer.wireBitfield.send({
+          start: i * this.core.bitfield.pageSize,
+          bitfield: p.bitfield
+        })
+      }
+    }
+
+    peer.protomux.uncork()
+  }
+
   async _onreorgdata (peer, req, data) {
     const f = this._addReorg(data.fork, peer)
 
@@ -1523,17 +1570,6 @@ module.exports = class Replicator {
   }
 }
 
-function pages (core) {
-  const res = []
-
-  for (let i = 0; i < core.tree.length; i += core.bitfield.pageSize) {
-    const p = core.bitfield.page(i / core.bitfield.pageSize)
-    res.push(p)
-  }
-
-  return res
-}
-
 function matchingRequest (req, data) {
   if (data.block !== null && (req.block === null || req.block.index !== data.block.index)) return false
   if (data.hash !== null && (req.hash === null || req.hash.index !== data.hash.index)) return false

test/replicate.js

@@ -781,3 +781,62 @@ test('non-sparse snapshot during partial replication', async function (t) {
   await s.update()
   t.is(s.contiguousLength, s.length)
 })
+
+test('sparse replication without gossiping', async function (t) {
+  t.plan(4)
+
+  const a = await create()
+  const b = await create(a.key)
+
+  await a.append(['a', 'b', 'c'])
+
+  let s
+
+  s = replicate(a, b)
+  await b.download({ start: 0, end: 3 }).downloaded()
+  await unreplicate(s)
+
+  await a.append(['d', 'e', 'f', 'd'])
+
+  s = replicate(a, b)
+  await b.download({ start: 4, end: 7 }).downloaded()
+  await unreplicate(s)
+
+  await t.test('block', async function (t) {
+    const c = await create(a.key)
+
+    s = replicate(b, c)
+    t.teardown(() => unreplicate(s))
+
+    t.alike(await c.get(4), Buffer.from('e'))
+  })
+
+  await t.test('range', async function (t) {
+    const c = await create(a.key)
+
+    s = replicate(b, c)
+    t.teardown(() => unreplicate(s))
+
+    await c.download({ start: 4, end: 6 }).downloaded()
+    t.pass('resolved')
+  })
+
+  await t.test('discrete range', async function (t) {
+    const c = await create(a.key)
+
+    s = replicate(b, c)
+    t.teardown(() => unreplicate(s))
+
+    await c.download({ blocks: [4, 6] }).downloaded()
+    t.pass('resolved')
+  })
+
+  await t.test('seek', async function (t) {
+    const c = await create(a.key)
+
+    s = replicate(b, c)
+    t.teardown(() => unreplicate(s))
+
+    t.alike(await c.seek(4), [4, 0])
+  })
+})