Add metric and test

Author: HDegroote

lib/replicator.js

@@ -371,7 +371,8 @@ class Peer {
       wireWant: { tx: 0, rx: 0 },
       wireBitfield: { tx: 0, rx: 0 },
       wireRange: { tx: 0, rx: 0 },
-      wireExtension: { tx: 0, rx: 0 }
+      wireExtension: { tx: 0, rx: 0 },
+      hotswaps: 0
     }
 
     this.receiverQueue = new ReceiverQueue()
@@ -1443,7 +1444,8 @@ module.exports = class Replicator {
       wireWant: { tx: 0, rx: 0 },
       wireBitfield: { tx: 0, rx: 0 },
       wireRange: { tx: 0, rx: 0 },
-      wireExtension: { tx: 0, rx: 0 }
+      wireExtension: { tx: 0, rx: 0 },
+      hotswaps: 0
     }
 
     this._attached = new Set()
@@ -2190,6 +2192,8 @@ module.exports = class Replicator {
 
     for (const b of this.hotswaps.pick(peer)) {
       if (peer._requestBlock(b) === false) continue
+      peer.stats.hotswaps++
+      peer.replicator.stats.hotswaps++
       if (peer.inflight >= maxHotswaps) break
     }
   }

package.json

@@ -66,6 +66,7 @@
   },
   "devDependencies": {
     "brittle": "^3.0.0",
+    "debugging-stream": "^3.1.0",
     "hyperswarm": "^4.3.6",
     "rache": "^1.0.0",
     "random-access-memory": "^6.1.0",

test/helpers/index.js

@@ -1,5 +1,6 @@
 const Hypercore = require('../../')
 const RAM = require('random-access-memory')
+const DebuggingStream = require('debugging-stream')
 
 exports.create = async function create (...args) {
   const core = new Hypercore(RAM, ...args)
@@ -56,6 +57,38 @@ exports.unreplicate = function unreplicate (streams) {
   }))
 }
 
+exports.replicateDebugStream = function replicate (a, b, t, opts = {}) {
+  const { latency, speed, jitter } = opts
+
+  const s1 = a.replicate(true, { keepAlive: false, ...opts })
+  const s2Base = b.replicate(false, { keepAlive: false, ...opts })
+  const s2 = new DebuggingStream(s2Base, { latency, speed, jitter })
+
+  s1.on('error', err => t.comment(`replication stream error (initiator): ${err}`))
+  s2.on('error', err => t.comment(`replication stream error (responder): ${err}`))
+
+  if (opts.teardown !== false) {
+    t.teardown(async function () {
+      let missing = 2
+      await new Promise(resolve => {
+        s1.on('close', onclose)
+        s1.destroy()
+
+        s2.on('close', onclose)
+        s2.destroy()
+
+        function onclose () {
+          if (--missing === 0) resolve()
+        }
+      })
+    })
+  }
+
+  s1.pipe(s2).pipe(s1)
+
+  return [s1, s2]
+}
+
 exports.eventFlush = async function eventFlush () {
   await new Promise(resolve => setImmediate(resolve))
 }

test/replicate.js

@@ -2,7 +2,7 @@ const test = require('brittle')
 const b4a = require('b4a')
 const RAM = require('random-access-memory')
 const NoiseSecretStream = require('@hyperswarm/secret-stream')
-const { create, replicate, unreplicate, eventFlush } = require('./helpers')
+const { create, replicate, unreplicate, eventFlush, replicateDebugStream } = require('./helpers')
 const { makeStreamPair } = require('./helpers/networking.js')
 const Hypercore = require('../')
 
@@ -51,9 +51,10 @@ test('basic replication stats', async function (t) {
   t.is(aStats.wireExtension.tx, 0, 'wireExtension init 0')
   t.is(aStats.wireCancel.rx, 0, 'wireCancel init 0')
   t.is(aStats.wireCancel.tx, 0, 'wireCancel init 0')
+  t.is(aStats.hotswaps, 0, 'hotswaps init 0')
 
   const initStatsLength = [...Object.keys(aStats)].length
-  t.is(initStatsLength, 8, 'Expected amount of stats')
+  t.is(initStatsLength, 9, 'Expected amount of stats')
 
   replicate(a, b, t)
 
@@ -1778,6 +1779,44 @@ test('replication count should never go negative', async function (t) {
   }
 })
 
+test('uses hotswaps to avoid long download tail', async t => {
+  const core = await create()
+  const slowCore = await create(core.key)
+
+  const batch = []
+  while (batch.length < 100) {
+    batch.push(Buffer.allocUnsafe(60000))
+  }
+  await core.append(batch)
+
+  replicate(core, slowCore, t)
+  await slowCore.download({ start: 0, end: core.length }).done()
+
+  t.is(slowCore.contiguousLength, 100, 'sanity check')
+
+  const peerCore = await create(core.key)
+  await peerCore.ready()
+  const [fastStream] = replicateDebugStream(core, peerCore, t, { speed: 10_000_000 })
+  const [slowStream] = replicateDebugStream(slowCore, peerCore, t, { speed: 1_000_000 })
+  const fastKey = fastStream.publicKey
+  const slowKey = slowStream.publicKey
+  const peerKey = fastStream.remotePublicKey
+  t.alike(peerKey, slowStream.remotePublicKey, 'sanity check')
+
+  await peerCore.download({ start: 0, end: core.length }).done()
+
+  const fastPeer = peerCore.replicator.peers.filter(
+    p => b4a.equals(p.stream.remotePublicKey, fastKey))[0]
+  const slowPeer = peerCore.replicator.peers.filter(
+    p => b4a.equals(p.stream.remotePublicKey, slowKey))[0]
+
+  t.ok(fastPeer.stats.hotswaps > 0, 'hotswaps happened for fast peer')
+  t.ok(fastPeer.stats.hotswaps > 0, 'No hotswaps happened for slow peer')
+  t.ok(slowPeer.stats.wireCancel.tx > 0, 'slow peer cancelled requests')
+  t.ok(fastPeer.stats.wireData.rx > slowPeer.stats.wireData.rx, 'sanity check: received more data from fast peer')
+  t.ok(slowPeer.stats.wireData.rx > 0, 'sanity check: still received data from slow peer')
+})
+
 async function waitForRequestBlock (core) {
   while (true) {
     const reqBlock = core.replicator._inflight._requests.find(req => req && req.block)