Delay updateAll() & only update subset immediately

index.js

@@ -719,7 +719,7 @@ class Hypercore extends EventEmitter {
       once = false
       this._findingPeers--
       if (this.core !== null && --this.core.replicator.findingPeers === 0) {
-        this.core.replicator.updateAll()
+        this.core.replicator.queueUpdateAll()
       }
     }
   }

lib/replicator.js

@@ -48,8 +48,8 @@ const SCALE_LATENCY = 50
 const NOT_DOWNLOADING_SLACK = (20000 + Math.random() * 20000) | 0
 const MAX_PEERS_UPGRADE = 3
 const LAST_BLOCKS = 256
-
 const MAX_RANGES = 64
+const MIN_PEER_UPDATE_ALL = 40
 
 const NOT_AVAILABLE = 1
 const INVALID_REQUEST = 2
@@ -753,7 +753,7 @@ class Peer {
 
     if (this.remoteOpened === false) {
       this.replicator._ifAvailable--
-      this.replicator.updateAll()
+      this.replicator.queueUpdateAll()
       return
     }
 
@@ -1297,7 +1297,7 @@ class Peer {
     const req = request > 0 ? this.replicator._inflight.get(request) : null
 
     if (req === null || req.peer !== this) {
-      this.replicator.updateAll()
+      this.replicator.queueUpdateAll()
       return
     }
 
@@ -2020,6 +2020,9 @@ module.exports = class Replicator {
     this._notDownloadingLinger = notDownloadingLinger
     this._notDownloadingTimer = null
 
+    this._updateAllBump = null
+    this._updateAllBound = this.updateAll.bind(this)
+
     const self = this
     this._onstreamclose = onstreamclose
 
@@ -2202,7 +2205,7 @@ module.exports = class Replicator {
 
     const ref = this._addUpgrade().attach(session)
 
-    this.updateAll()
+    this.queueUpdateAll()
 
     return ref
   }
@@ -2212,7 +2215,7 @@ module.exports = class Replicator {
     const ref = b.attach(session)
 
     this._queueBlock(b)
-    this.updateAll()
+    this.queueUpdateAll()
 
     return ref
   }
@@ -2222,7 +2225,7 @@ module.exports = class Replicator {
     const ref = s.attach(session)
 
     this._seeks.push(s)
-    this.updateAll()
+    this.queueUpdateAll()
 
     return ref
   }
@@ -2257,7 +2260,7 @@ module.exports = class Replicator {
       return ref
     }
 
-    this.updateAll()
+    this.queueUpdateAll()
 
     return ref
   }
@@ -2279,7 +2282,7 @@ module.exports = class Replicator {
     }
 
     for (const replicator of updated) {
-      replicator.updateAll()
+      replicator.queueUpdateAll()
     }
   }
 
@@ -2291,7 +2294,7 @@ module.exports = class Replicator {
       cleared = true
     }
 
-    if (cleared) this.updateAll()
+    if (cleared) this.queueUpdateAll()
   }
 
   _matchingRequest(req, data) {
@@ -2452,7 +2455,7 @@ module.exports = class Replicator {
 
     this._onpeerupdate(false, peer)
     if (inflight) {
-      this.updateAll()
+      this.queueUpdateAll()
     } else {
       this._checkUpgradeIfAvailable()
     }
@@ -2645,7 +2648,7 @@ module.exports = class Replicator {
       this._updatesPending = 0
     }
 
-    if (this._inflight.idle || updateAll) this.updateAll()
+    if (this._inflight.idle || updateAll) this.queueUpdateAll()
   }
 
   _clearRequest(peer, req) {
@@ -2694,7 +2697,7 @@ module.exports = class Replicator {
       this._clearRequest(peer, req)
     }
 
-    this.updateAll()
+    this.queueUpdateAll()
   }
 
   _openSkipBitfield() {
@@ -2817,7 +2820,7 @@ module.exports = class Replicator {
     const f = this._addReorg(data.fork, peer)
 
     if (f === null) {
-      this.updateAll()
+      this.queueUpdateAll()
       return
     }
 
@@ -2838,7 +2841,7 @@ module.exports = class Replicator {
       }
     }
 
-    this.updateAll()
+    this.queueUpdateAll()
   }
 
   // Never throws, allowed to run in the background
@@ -2878,7 +2881,7 @@ module.exports = class Replicator {
     for (const f of old) f.resolve()
     f.resolve()
 
-    this.updateAll()
+    this.queueUpdateAll()
   }
 
   _maybeUpdate() {
@@ -3000,16 +3003,30 @@ module.exports = class Replicator {
     this._checkUpgradeIfAvailable()
   }
 
-  updateAll() {
+  updateSubset() {
+    return this.updateAll(false)
+  }
+
+  updateAll(all = true) {
+    // Clear queued scan if a full scan
+    if (this._updateAllBump !== null && all) {
+      clearTimeout(this._updateAllBump)
+      this._updateAllBump = null
+    }
+
     // Quick shortcut to wait for flushing reorgs - not needed but less waisted requests
     if (this._applyingReorg !== null) return
 
     const peers = new RandomIterator(this.peers)
 
+    const limit = all ? peers.length : MIN_PEER_UPDATE_ALL
+    let tried = 0
     for (const peer of peers) {
+      tried++
       if (this._updatePeer(peer) === true) {
         peers.requeue()
       }
+      if (tried >= limit) break
     }
 
     // Check if we can skip the non primary check fully
@@ -3018,15 +3035,33 @@ module.exports = class Replicator {
       return
     }
 
+    tried = 0
     for (const peer of peers.restart()) {
+      tried++
       if (this._updatePeerNonPrimary(peer) === true) {
         peers.requeue()
       }
+      if (tried >= limit) break
     }
 
     this._checkUpgradeIfAvailable()
   }
 
+  getUpdateAllDelay(peers) {
+    return Math.min(3000, Math.max(100, (peers.length - MIN_PEER_UPDATE_ALL) * 5))
+  }
+
+  queueUpdateAll() {
+    if (this.peers.length < MIN_PEER_UPDATE_ALL) return this.updateAll()
+
+    // Immediately scan subset
+    this.updateSubset()
+
+    // Schedule full scan
+    if (this._updateAllBump !== null) return //skip if already scheduled
+    this._updateAllBump = setTimeout(this._updateAllBound, this.getUpdateAllDelay(this.peers))
+  }
+
   onpeerdestroy() {
     if (--this._active === 0) this.core.checkIfIdle()
   }
@@ -3090,6 +3125,11 @@ module.exports = class Replicator {
       this._notDownloadingTimer = null
     }
 
+    if (this._updateAllBump) {
+      clearTimeout(this._updateAllBump)
+      this._updateAllBump = null
+    }
+
     while (this.peers.length) {
       const peer = this.peers[this.peers.length - 1]
       this.detachFrom(peer.protomux)

test/replicate.js

@@ -2830,6 +2830,67 @@ test('backoff respected for NOT_AVAILABLE in case of incorrectly false remote bi
   t.ok(reader.replicator.stats.wireRequest.tx < 40, 'did not continue spamming (sanity check)')
 })
 
+test('delayed updateAll clears timer after reorg skip', async function (t) {
+  const core = await create(t)
+
+  const targetPeers = 40
+  const gotAllPeers = new Promise((resolve) => {
+    core.on('peer-add', () => {
+      if (core.replicator.peers.length === targetPeers) resolve()
+    })
+  })
+  for (let i = 0; i < targetPeers; i++) {
+    const peer = await create(t, core.key)
+    replicate(core, peer, t)
+  }
+
+  await gotAllPeers
+
+  const r = core.replicator
+
+  t.is(r.peers.length, targetPeers, 'has minimum peers to defer full scan')
+
+  // Simulate the delayed full scan firing while a reorg is active
+  // TODO create a more realistic simulation
+  r._applyingReorg = {}
+  r.queueUpdateAll()
+
+  // wait minimum amount to clear timeout for full scan (min is 100)
+  await new Promise((resolve) => setTimeout(resolve, 120))
+
+  t.is(r._updateAllBump, null, 'timer handle should be cleared after firing')
+})
+
+test('delayed updateAll timer doesnt keep event loop alive', async function (t) {
+  const core = await create(t)
+
+  const targetPeers = 40
+  const gotAllPeers = new Promise((resolve) => {
+    core.on('peer-add', () => {
+      if (core.replicator.peers.length === targetPeers) resolve()
+    })
+  })
+  for (let i = 0; i < targetPeers; i++) {
+    create(t, core.key).then((peer) => replicate(core, peer, t))
+  }
+
+  await gotAllPeers
+
+  const r = core.replicator
+
+  t.is(r.peers.length, targetPeers, 'has minimum peers to defer full scan')
+
+  // Simulate the delayed full scan firing while a reorg is active
+  // TODO create a more realistic simulation
+  r._applyingReorg = {}
+  r.queueUpdateAll()
+  t.not(r._updateAllBump, null, 'timer set')
+
+  r.destroy()
+  t.ok(r.destroyed, 'replicator destroyed')
+  t.is(r._updateAllBump, null, 'timer reset to null')
+})
+
 async function createAndDownload(t, core) {
   const b = await create(t, core.key)
   replicate(core, b, t, { teardown: false })