Rebuild: prefetching no longer waits for merging, nor stalls on finished ranges

The coordinator ran the prefetch tick in the same loop as the merger
(up to 30 s a minute): its connections went idle once their targets
were fetched, and the first leases after a start found nothing spooled.
The prefetcher now ticks in a loop of its own.

The prefetcher took the first planned ranges with nothing spooled, even
ones with nothing left to fetch (next_id past the end): two such ranges
(edge:bench's, set back to planned) kept it from fetching anything. It
and the lease query now skip those, and edge:bench marks its range done.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
Daan MeijerandClaude Opus 5.5 committed 2026-10-09 03:04:00 +02:00
1 parent 2b4f673ddb
commit bd1c846f27
4 files changed
+18 -4

No files matched your search

+1 -1
View File
@@ -19,7 +19,7 @@ class Leases {
`UPDATE ranges r SET status = 'leased', node_id = ?, leased_at = now(), heartbeat_at = now(), attempts = attempts + 1, rule_version = ?
WHERE r.id = (
SELECT id FROM ranges
WHERE spooled_chunks > 0 AND (status = 'planned' OR (status = 'leased' AND heartbeat_at < now() - make_interval(secs => ?)))
WHERE spooled_chunks > 0 AND next_id <= end_id AND (status = 'planned' OR (status = 'leased' AND heartbeat_at < now() - make_interval(secs => ?)))
ORDER BY priority, id LIMIT 1 FOR UPDATE SKIP LOCKED)
RETURNING r.id, r.group_id, r.start_id, r.end_id, r.next_id`,
[nodeId, rules.version, coordinatorConfig.leaseTtlSeconds]
+1 -1
View File
@@ -114,7 +114,7 @@ export class Prefetcher {
const [{ ready }] = await db.from('ranges').where('status', 'planned').where('spooled_chunks', '>', 0).count('* as ready')
const room = this.spoolBytes < coordinatorConfig.spool.maxBytes
const fresh = room && Number(ready) < ahead
? await rows(db.query()).where('r.status', 'planned').where('r.spooled_chunks', 0).orderBy([{ column: 'r.priority' }, { column: 'r.id' }]).limit(ahead - Number(ready))
? await rows(db.query()).where('r.status', 'planned').where('r.spooled_chunks', 0).whereRaw('r.next_id <= r.end_id').orderBy([{ column: 'r.priority' }, { column: 'r.id' }]).limit(ahead - Number(ready))
: []
const targets: Target[] = []
for (const row of [...leased, ...(room ? started : []), ...fresh]) {
+14 -2
View File
@@ -34,9 +34,21 @@ export default class CoordinatorRun extends BaseCommand {
prefetcher.stop()
await done
})
this.logger.info('Coordinator started.')
// Two loops: the prefetcher's ticks don't wait for merging or planning (a merge round
// of up to MERGE_BUDGET_MS left the connections idle once their targets were fetched).
const prefetching = (async () => {
while (!stopping) {
try {
await prefetcher.tick()
} catch (error: any) {
this.logger.error(`Prefetch round failed: ${error.message}`)
}
await sleep(TICK_MS)
}
})()
let plannedAt = 0
let mergedAt = 0
this.logger.info('Coordinator started.')
while (!stopping) {
try {
if (Date.now() - plannedAt >= PLAN_EVERY_MS) {
@@ -65,12 +77,12 @@ export default class CoordinatorRun extends BaseCommand {
}
if (totals.completed || totals.incomplete) this.logger.info(`Fragments: ${totals.completed} files completed, ${totals.incomplete} incomplete, ${totals.waiting} waiting.`)
}
await prefetcher.tick()
} catch (error: any) {
this.logger.error(`Coordinator round failed: ${error.message}`)
}
await sleep(TICK_MS)
}
await prefetching
while (prefetcher.active > 0) await sleep(200)
await NntpPool.close()
stopped!()
+2
View File
@@ -163,6 +163,8 @@ export default class EdgeBench extends BaseCommand {
peakDetail = `heap ${(m.heapUsed / 1e6).toFixed(0)}/${(m.heapTotal / 1e6).toFixed(0)} MB, external ${(m.external / 1e6).toFixed(0)} MB, array buffers ${(m.arrayBuffers / 1e6).toFixed(0)} MB`
}
}
// Done, so nothing takes it for a range to work on (a leased one was set back to planned once).
if (this.store) await db.from('ranges').where('id', rangeId).update({ status: 'done', done_at: new Date() })
const cpu = process.cpuUsage(cpuStart)
const cpuS = (cpu.user + cpu.system) / 1e6
const total = t.inflate + t.parse + t.index