Fix idle range completion scan

lib/replicator.js

@@ -2600,23 +2600,60 @@ module.exports = class Replicator {
     }
   }
 
+  _updateRanges(index, limit) {
+    if (this._ranges.length === 0) {
+      return { checked: 0, resolved: 0, index: 0 }
+    }
+
+    index = Math.min(index, this._ranges.length - 1)
+
+    let checked = 0
+    let resolved = 0
+    let remaining = Math.min(limit, this._ranges.length)
+
+    while (remaining-- > 0 && this._ranges.length > 0) {
+      if (index >= this._ranges.length) index = 0
+
+      const r = this._ranges[index]
+
+      clampRange(this.core, r)
+      checked++
+
+      if (r.end !== -1 && r.start >= r.end) {
+        this._resolveRangeRequest(r)
+        resolved++
+      } else {
+        index++
+      }
+    }
+
+    index = this._ranges.length === 0 ? 0 : index % this._ranges.length
+
+    return { checked, resolved, index }
+  }
+
   // "slow" updates here - async but not allowed to ever throw
   async _updateNonPrimary(updateAll) {
     // Check if running, if so skip it and the running one will issue another update for us (debounce)
     while (++this._updatesPending === 1) {
-      let len = Math.min(MAX_RANGES, this._ranges.length)
+      let checkedSinceResolve = 0
+      let rangeIndex = 0
+      const drain = this._inflight.idle || updateAll
 
-      for (let i = 0; i < len; i++) {
-        const r = this._ranges[i]
+      while (this._ranges.length > 0) {
+        const limit = Math.min(MAX_RANGES, this._ranges.length - checkedSinceResolve)
+        const result = this._updateRanges(rangeIndex, limit)
+        const { checked, resolved } = result
+        rangeIndex = result.index
 
-        clampRange(this.core, r)
+        if (resolved > 0) checkedSinceResolve = 0
+        else checkedSinceResolve += checked
 
-        if (r.end !== -1 && r.start >= r.end) {
-          this._resolveRangeRequest(r)
-          i--
-          if (len > this._ranges.length) len--
-          if (this._ranges.length === MAX_RANGES) updateAll = true
-        }
+        if (checked === 0 || checkedSinceResolve >= this._ranges.length) break
+        if (!drain) break
+
+        await yieldToLoop()
+        if (this.destroyed) break
       }
 
       for (let i = 0; i < this._seeks.length; i++) {
@@ -3384,6 +3421,10 @@ function incrementRx(stats1, stats2) {
 
 function noop() {}
 
+function yieldToLoop() {
+  return new Promise((resolve) => setTimeout(resolve, 0))
+}
+
 function backoff(times) {
   const sleep = times < 2 ? 200 : times < 5 ? 500 : times < 40 ? 1000 : 5000
   return new Promise((resolve) => setTimeout(resolve, sleep))

test/replicate.js

@@ -2891,6 +2891,119 @@ test('delayed updateAll timer doesnt keep event loop alive', async function (t)
   t.is(r._updateAllBump, null, 'timer reset to null')
 })
 
+test('idle range completion drains past max range window', async function (t) {
+  const core = await create(t)
+  const clone = await create(t, core.key)
+
+  const pending = new Set()
+  const complete = []
+  const totalLength = 150
+  const availableStart = 100
+
+  for (let i = 0; i < totalLength; i++) {
+    await core.append('' + i)
+  }
+
+  await core.clear(0, availableStart)
+
+  const random = Math.random
+  Math.random = () => 0.999
+  t.teardown(() => {
+    Math.random = random
+  })
+
+  replicate(core, clone, t)
+
+  for (let i = 0; i < totalLength; i++) {
+    pending.add(i)
+
+    const done = clone.download({ start: i, end: i + 1 }).done()
+    done.then(() => pending.delete(i), noop)
+
+    if (i >= availableStart) complete.push(done)
+  }
+
+  t.is(clone.core.replicator._ranges.length, totalLength, 'all ranges are pending')
+  t.ok(await core.has(availableStart, totalLength), 'source has all available blocks')
+
+  const resolved = await Promise.race([
+    Promise.all(complete).then(() => true),
+    new Promise((resolve) => setTimeout(resolve, 1000, false))
+  ])
+
+  t.ok(resolved, 'all locally complete ranges resolved')
+  t.is(pending.size, availableStart, 'only unavailable ranges remain pending')
+  t.is(clone.core.replicator._ranges.length, availableStart, 'resolved ranges were removed')
+
+  const outliers = []
+  for (const range of clone.core.replicator._ranges) {
+    if (range.userStart >= availableStart) outliers.push(range.userStart)
+  }
+  t.alike(outliers, [], 'no available ranges remain pending')
+})
+
+test('idle range completion keeps draining if update queues during yield', async function (t) {
+  const core = await create(t)
+  const pending = new Set()
+  const complete = []
+
+  const totalLength = 150
+  const availableStart = 100
+  const replicator = core.core.replicator
+
+  for (let i = 0; i < totalLength; i++) {
+    pending.add(i)
+
+    const done = core.download({ start: i, end: i + 1 }).done()
+    done.then(() => pending.delete(i), noop)
+
+    if (i >= availableStart) complete.push(done)
+  }
+
+  t.is(replicator._ranges.length, totalLength, 'all ranges are pending')
+
+  core.core._setBitfieldRanges(availableStart, totalLength, true)
+
+  const updateRanges = replicator._updateRanges
+  let updates = 0
+
+  replicator._updateRanges = function (index, limit) {
+    const result = updateRanges.call(this, index, limit)
+
+    if (++updates === 1) {
+      setTimeout(() => {
+        this._inflight._active++
+        this._updateNonPrimary(false).catch(noop)
+      }, 0)
+    }
+
+    return result
+  }
+
+  t.teardown(() => {
+    replicator._updateRanges = updateRanges
+    replicator._inflight._active = 0
+  })
+
+  await replicator._updateNonPrimary(false)
+  replicator._inflight._active = 0
+
+  const resolved = await Promise.race([
+    Promise.all(complete).then(() => true),
+    new Promise((resolve) => setTimeout(resolve, 1000, false))
+  ])
+
+  t.ok(resolved, 'all locally complete ranges resolved')
+  t.is(pending.size, availableStart, 'only unavailable ranges remain pending')
+  t.is(replicator._ranges.length, availableStart, 'resolved ranges were removed')
+
+  const outliers = []
+  for (const range of replicator._ranges) {
+    if (range.userStart >= availableStart) outliers.push(range.userStart)
+  }
+  t.alike(outliers, [], 'no available ranges remain pending')
+})
+
 async function createAndDownload(t, core) {
   const b = await create(t, core.key)
   replicate(core, b, t, { teardown: false })