Skip to content

Fix idle range completion scan #832

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
marcus-pousette-hp wants to merge 13 commits into
base: main
Choose a base branch
from
+448 −36
Open

Fix idle range completion scan #832

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
13 commits
Select commit Hold shift + click to select a range
f9700d5
Fix idle range completion scan
marcus-pousette-hp Jun 22, 2026
1414e16
Fix stuck range completion drain
marcus-pousette-hp Jun 23, 2026
76376ac
Remove unnecessary range update boundary state
marcus-pousette-hp Jun 23, 2026
60823b1
Fix queued idle range drain race
marcus-pousette-hp Jun 23, 2026
5403eb5
Restore peer updates at max range boundary
marcus-pousette-hp Jul 13, 2026
37224a9
Yield range drains to pending seeks
marcus-pousette-hp Jul 14, 2026
ffd7033
Assert seeks precede range scan completion
marcus-pousette-hp Jul 14, 2026
fe789a1
Restart range drain after cancellation
marcus-pousette-hp Jul 14, 2026
62096ed
Simplify range drain state and tests
marcus-pousette-hp Jul 15, 2026
aded7ec
Remove redundant range empty checks
marcus-pousette-hp Jul 15, 2026
11e0082
Merge main into fix/idle-range-completion
marcus-pousette-hp Jul 16, 2026
0e24e5a
Fix seek processing during range drains
marcus-pousette-hp Jul 16, 2026
51738b8
Clear range test timeout
marcus-pousette-hp Jul 16, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
142 changes: 106 additions & 36 deletions lib/replicator.js
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -285,6 +285,10 @@ class RangeRequest extends Attachable {
this.ranges[rangeIndex] = h
}

if (!this.resolved && this.replicator._updatesRunning) {
this.replicator._updatesQueued = true
}

if (this.end === -1) {
this.replicator._alwaysLatestBlock--
}
Expand Down Expand Up @@ -1541,8 +1545,8 @@ class Peer {
}

_requestSeek(s) {
// if replicator is updating the seeks etc, bail and wait for it to drain
if (this.replicator._updatesPending > 0) return false
// if replicator is updating the seeks, bail and wait for it to drain
if (this.replicator._updatingSeeks) return false
if (this.replicator.pushOnly) return false

const { length, fork } = this.core.state
Expand Down Expand Up @@ -2028,7 +2032,9 @@ module.exports = class Replicator {
this._hadPeers = false
this._active = 0
this._ifAvailable = 0
this._updatesPending = 0
this._updatingSeeks = false
this._updatesRunning = false
this._updatesQueued = false
this._applyingReorg = null
this._manifestPeer = null
this._notDownloadingLinger = notDownloadingLinger
Expand Down Expand Up @@ -2614,55 +2620,115 @@ module.exports = class Replicator {
}
}

_updateRanges(index, limit) {
index = Math.min(index, this._ranges.length - 1)

let checked = 0
let resolved = 0
let updateAll = false
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++
// Crossing into the capped window exposes another range to peer scheduling.
if (this._ranges.length === MAX_RANGES) updateAll = true
} else {
index++
}
}

index = this._ranges.length === 0 ? 0 : index % this._ranges.length

return { checked, resolved, index, updateAll }
}

// "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)
this._updatesQueued = true
if (this._updatesRunning) {
if (this._inflight.idle || updateAll) this.queueUpdateAll()
return
}

for (let i = 0; i < len; i++) {
const r = this._ranges[i]
this._updatesRunning = true

clampRange(this.core, r)
while (this._updatesQueued) {
this._updatesQueued = false

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

@marcus-pousette-hp marcus-pousette-hp Jun 23, 2026

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Removed the old this._ranges.length === MAX_RANGES special case because it only re-triggered updateAll when the queue shrank to exactly the cap, and I couldn’t find test coverage or PR discussion showing that boundary behavior is intentional.

@marcus-pousette-hp marcus-pousette-hp Jun 23, 2026

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

The actual MAX_RANGES cap is still preserved. The new cursor loop drains completed ranges in capped chunks without relying on that exact effect.

@lejeunerenard lejeunerenard Jul 9, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I dont think this should be removed. The updateAll prompts requesting ranges from a random subset of peers potentially. Since the _updateNonPrimary function is called only when either receiving data or an upgrade, it could deadlock with requests queue in _ranges that haven't been requested of peers that can fulfill them if updateAll() isn't triggered.

The reason for the exact match is that we know in this scenario only 1 range was resolved so this check triggers as soon as there can be at least one range that could not have been requested of peers. Aka triggering another set of range requests could prompt a response which resolves a range not included in the last _updateNonPrimary run.

This doesn't prevent from large backlogs not getting stuck (hence the need for this PR) when there is more unfulfilled range request than the MAX_RANGES though. I don't think it should be updateAll = true for this._ranges.length >= MAX_RANGES though.

@marcus-pousette-hp marcus-pousette-hp Jul 13, 2026

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Ye, that make sense. I restored this in 5403eb5, preserving the exact this._ranges.length === MAX_RANGES boundary. I also added a regression test with 65 one-block ranges and two peers as an proof/argumetn for this. Though it is quite verbose right now, I will re-iterate on this tomorrow.

}
}
let checkedSinceResolve = 0
let rangeIndex = 0
const drain = this._inflight.idle || updateAll
const checkedSeeks = new Set()

for (let i = 0; i < this._seeks.length; i++) {
const s = this._seeks[i]
while (true) {
const limit = Math.min(MAX_RANGES, this._ranges.length - checkedSinceResolve)
const result = this._updateRanges(rangeIndex, limit)
const { checked, resolved } = result
rangeIndex = result.index

let err = null
let res = null
if (result.updateAll) updateAll = true
Comment on lines +2674 to +2678

@lejeunerenard lejeunerenard Jul 21, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Suggested change
const result = this._updateRanges(rangeIndex, limit)
const { checked, resolved } = result
rangeIndex = result.index
let err = null
let res = null
if (result.updateAll) updateAll = true
const { checked, resolved, index, updateAll: resultUpdateAll } = this._updateRanges(rangeIndex, limit)
rangeIndex = index
if (resultUpdateAll) updateAll = true


try {
res = await s.seeker.update()
} catch (error) {
err = error
}
if (resolved > 0) checkedSinceResolve = 0
else checkedSinceResolve += checked

const continueRanges = checked > 0 && checkedSinceResolve < this._ranges.length && drain

this._updatingSeeks = true

let checkedSeek = false
for (const s of this._seeks.slice()) {

@lejeunerenard lejeunerenard Jul 21, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Just a note as I dont think it is an issue in practice (because seeks are mainly used for streaming), but there is no limit to the number of seeks processed. This means just like ranges held up seek resolution, the opposite is still true where a large number of seeks could hold up range updates (allowing the initial update of a batch of ranges).

I propose we limit the length as well and use RandomIterator. Here is a test I made to showcase the issue:

test('seeks cant block range updates', async function (t) {
  const a = await create(t)
  const b = await create(t, a.key)
  const totalBlocks = 1e5
  const totalRanges = 129 // above max range for at least 2x batches
  const totalSeeks = totalBlocks

  // Seed data
  await a.append(Array(totalBlocks).fill('a'))
  await a.clear(0, a.length - totalRanges) // Clear all but the end

  replicate(a, b, t)

  // Setup ranges at the end
  const ranges = []
  for (let i = 0; i < totalRanges; i++) {
    ranges.push(b.download({ start: a.length - (i + 1), end: a.length - i }).done())
  }

  // Setup large number of seeks
  const seeks = []
  for (let i = 0; i < totalSeeks; i++) {
    seeks.push(b.seek(i).catch(noop))
  }

  // Allow seeks to be added to the replicator
  await eventFlush()

  const result = await Promise.race([
    b.core.replicator._updateNonPrimary(true).then(() => 'resolved'),
    new Promise((resolve) => setTimeout(resolve, 100)).then(() => 'timeout')
  ])
  t.is(result, 'resolved', 'updating non-primary didnt block on seeks')
})

if (checkedSeeks.has(s) || !this._seeks.includes(s)) continue
checkedSeeks.add(s)
checkedSeek = true

let err = null
let res = null

try {
res = await s.seeker.update()
} catch (error) {
err = error
}

if (!res && !err) continue
const seekIndex = this._seeks.indexOf(s)
if (seekIndex === -1 || (!res && !err)) continue

if (i < this._seeks.length - 1) this._seeks[i] = this._seeks.pop()
else this._seeks.pop()
if (seekIndex < this._seeks.length - 1) {
this._seeks[seekIndex] = this._seeks.pop()
} else this._seeks.pop()
Comment on lines +2705 to +2707

@lejeunerenard lejeunerenard Jul 21, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This can be slightly simpler if we always pop and only swap if the popped version isn't the s. Something like this:

hypercore/lib/replicator.js

Lines 283 to 286 in 3cd8c50

const h = this.ranges.pop()
if (h !== this) {
this.ranges[rangeIndex] = h
}


i--
if (err) s.reject(err)
else s.resolve(res)
}

this._updatingSeeks = false

if (
(!continueRanges || (checkedSeek && this._seeks.length > 0)) &&
(this._inflight.idle || updateAll)
) {
this.queueUpdateAll()
}
if (!continueRanges) break

if (err) s.reject(err)
else s.resolve(res)
await yieldToLoop()
if (this.destroyed) break
}

// No additional updates scheduled - break
if (--this._updatesPending === 0) break
// Debounce the additional updates - continue
this._updatesPending = 0
if (this.destroyed) break
}

if (this._inflight.idle || updateAll) this.queueUpdateAll()
this._updatingSeeks = false
this._updatesRunning = false
}

_clearRequest(peer, req) {
Expand Down Expand Up @@ -3398,6 +3464,10 @@ function incrementRx(stats1, stats2) {

function noop() {}

function yieldToLoop() {
return new Promise((resolve) => setImmediate(resolve))
}

function backoff(times) {
const sleep = times < 2 ? 200 : times < 5 ? 500 : times < 40 ? 1000 : 5000
return new Promise((resolve) => setTimeout(resolve, sleep))
Expand Down
Loading