From bd1c846f27f8423bd71a060f2493a53502960544 Mon Sep 17 00:00:00 2001 From: Daan Meijer Date: Fri, 9 Oct 2026 03:04:00 +0200 Subject: [PATCH] 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 --- app/services/leases.ts | 2 +- app/services/prefetcher.ts | 2 +- commands/coordinator_run.ts | 16 ++++++++++++++-- commands/edge_bench.ts | 2 ++ 4 files changed, 18 insertions(+), 4 deletions(-) diff --git a/app/services/leases.ts b/app/services/leases.ts index 2bb8f40..9a9b385 100644 --- a/app/services/leases.ts +++ b/app/services/leases.ts @@ -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] diff --git a/app/services/prefetcher.ts b/app/services/prefetcher.ts index a7a2c45..1d93881 100644 --- a/app/services/prefetcher.ts +++ b/app/services/prefetcher.ts @@ -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]) { diff --git a/commands/coordinator_run.ts b/commands/coordinator_run.ts index 097e5a0..6e06e3c 100644 --- a/commands/coordinator_run.ts +++ b/commands/coordinator_run.ts @@ -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!() diff --git a/commands/edge_bench.ts b/commands/edge_bench.ts index 423ce0f..43489f8 100644 --- a/commands/edge_bench.ts +++ b/commands/edge_bench.ts @@ -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