From 7b436f3f7754a48597a9598865de4217b4599beb Mon Sep 17 00:00:00 2001 From: Daan Meijer Date: Mon, 12 Oct 2026 20:11:57 +0200 Subject: [PATCH] Idle connections fetch capture ranges into the corpus The spool only fetches what workers lease soon, in priority order, and the tail never ends: after a reset the 1,400 capture ranges waited behind it while 34 of 40 connections idled. Up to CORPUS_AHEAD (8) planned capture ranges whose chunks aren't all stored are now targets after everything else, fetched into the corpus only (no spool); a later lease reads them from disk. Co-Authored-By: Claude Opus 5.5 --- app/services/corpus.ts | 20 ++++++++++++- app/services/prefetcher.ts | 45 ++++++++++++++++++++++++++++-- config/coordinator.ts | 6 ++++ start/env.ts | 1 + tests/unit/services/corpus.spec.ts | 37 ++++++++++++++++++++++++ 5 files changed, 106 insertions(+), 3 deletions(-) create mode 100644 tests/unit/services/corpus.spec.ts diff --git a/app/services/corpus.ts b/app/services/corpus.ts index a5577de..bb636b8 100644 --- a/app/services/corpus.ts +++ b/app/services/corpus.ts @@ -1,4 +1,4 @@ -import { mkdir, readFile, readdir, rename, rm, writeFile } from 'node:fs/promises' +import { access, mkdir, readFile, readdir, rename, rm, writeFile } from 'node:fs/promises' import { join } from 'node:path' import { unzip } from 'node:zlib' import { promisify } from 'node:util' @@ -35,6 +35,24 @@ class Corpus { return join(this.rangeDir(range), `${String(index).padStart(4, '0')}-${span.from}-${span.to}.gz`) } + /** Whether every chunk of the range is stored (its range.json). */ + async complete(range: CorpusRange): Promise { + if (!this.dir) return false + return access(join(this.rangeDir(range), 'range.json')).then( + () => true, + () => false + ) + } + + /** Whether one chunk is stored. */ + async has(range: CorpusRange, span: ChunkSpan): Promise { + if (!this.dir) return false + return access(this.path(range, span)).then( + () => true, + () => false + ) + } + /** A stored chunk, or null. */ async read(range: CorpusRange, span: ChunkSpan): Promise { if (!this.dir) return null diff --git a/app/services/prefetcher.ts b/app/services/prefetcher.ts index f7892b1..d88a1b3 100644 --- a/app/services/prefetcher.ts +++ b/app/services/prefetcher.ts @@ -19,6 +19,8 @@ interface Target { end: number /** Its chunks are kept in the corpus (ranges.capture), and read from there when stored. */ capture: boolean + /** Fetched into the corpus only, not the spool: a capture range no worker leases soon (corpus.ahead). */ + corpusOnly?: boolean } /** A request's bytes in are sampled this often; STALL_SAMPLES of them make the minute a stall is judged over. */ @@ -70,12 +72,16 @@ export class Prefetcher { private sampler?: NodeJS.Timeout /** Requests in flight: what, since when, on which connection (the report names the oldest). */ private inFlight = new Set<{ rangeId: number; from: number; to: number; since: number; client: XoverClient }>() + /** Capture ranges found complete in the corpus (not looked at again). */ + private stored = new Set() private freshStats() { return { chunks: 0, /** Chunks read from the corpus instead of fetched. */ stored: 0, + /** Chunks fetched into the corpus only (corpus.ahead). */ + captured: 0, articles: 0, fetchMs: 0, hedged: 0, @@ -150,7 +156,7 @@ export class Prefetcher { } const corpusRange = { group: target.group, start: target.start, end: target.end } // A range in the corpus is read from disk (a reset re-indexes it without the news server). - let data = target.capture ? await Corpus.read(corpusRange, span) : null + let data = target.capture && !target.corpusOnly ? await Corpus.read(corpusRange, span) : null const stored = data !== null try { if (!data) data = target.plain.has(span.from) ? await NntpPool.usePlain(fetch) : await this.fetchHedged(target.id, target.group, span) @@ -165,6 +171,10 @@ export class Prefetcher { this.stats.fetchMs += Date.now() - fetchedAt if (target.capture) await Corpus.write(corpusRange, span, data, target.total) } + if (target.corpusOnly) { + this.stats.captured++ + continue + } const writtenAt = Date.now() this.writingNow++ try { @@ -304,7 +314,9 @@ export class Prefetcher { `Prefetched ${chunks} chunks in ${seconds.toFixed(0)} s: ${Math.round(articles / seconds).toLocaleString('en')} articles/s, ` + `${(fetchMs / chunks / 1000).toFixed(1)} s a chunk; ` + `hedged ${hedged} (second won ${hedgeWon}); slow by connection age: ${ages}` + - (this.stats.stored > 0 ? `; ${this.stats.stored} chunks from the corpus.` : '.') + (this.stats.stored > 0 ? `; ${this.stats.stored} chunks from the corpus` : '') + + (this.stats.captured > 0 ? `; ${this.stats.captured} fetched into the corpus only` : '') + + '.' ) const st = this.stats const avg = (n: number) => (samples > 0 ? n / samples : 0) @@ -390,6 +402,35 @@ export class Prefetcher { targets.push({ id: Number(row.id), group: row.group, plain: new Set((row.plain_chunks ?? []).map(Number)), missing, total, start, end, capture: row.capture === true }) } } + // Last, for loops with nothing else to do: capture ranges not stored yet, into the corpus only. + for (const target of await this.corpusTargets(rows, new Set(targets.map((t) => t.id)))) targets.push(target) + return targets + } + + /** Up to corpus.ahead planned capture ranges, not leased soon, whose chunks aren't all in the corpus. */ + private async corpusTargets(rows: (q: any) => any, taken: Set): Promise { + const { dir, ahead } = coordinatorConfig.corpus + if (!dir || ahead <= 0) return [] + const candidates = await rows(db.query()).where('r.status', 'planned').where('r.capture', true).where('r.spooled_chunks', 0).orderBy('r.id') + const targets: Target[] = [] + for (const row of candidates) { + if (targets.length >= ahead) break + const id = Number(row.id) + if (taken.has(id) || this.stored.has(id)) continue + const start = Number(row.start_id) + const end = Number(row.end_id) + const range = { group: row.group, start, end } + if (await Corpus.complete(range)) { + this.stored.add(id) + continue + } + const chunks = Spool.chunksOf({ start, end }) + const missing: ChunkSpan[] = [] + for (const span of chunks) if (!this.claimed.has(`${id}:${span.from}`) && !(await Corpus.has(range, span))) missing.push(span) + if (missing.length > 0) { + targets.push({ id, group: row.group, plain: new Set((row.plain_chunks ?? []).map(Number)), missing, total: chunks.length, start, end, capture: true, corpusOnly: true }) + } + } return targets } } diff --git a/config/coordinator.ts b/config/coordinator.ts index c3528ab..91888ab 100644 --- a/config/coordinator.ts +++ b/config/coordinator.ts @@ -80,6 +80,12 @@ export default { /** Stored XOVER chunks of ranges marked `capture`, for the grouping sweeps (app/services/corpus.ts). Off when unset. */ corpus: { dir: env.get('CORPUS_DIR'), + /** + * Capture ranges fetched into the corpus by connections the spool leaves idle, this many at a time: the + * spool only fetches what workers lease soon, and the tail comes first, so after a reset the capture + * ranges waited behind it for hours (1,400 marked, none stored, 34 of 40 connections idle; LOG.md 2026-10-12 (27)). + */ + ahead: env.get('CORPUS_AHEAD', 8), }, spool: { dir: env.get('SPOOL_DIR', app.tmpPath('spool')), diff --git a/start/env.ts b/start/env.ts index 7667e86..3754d48 100644 --- a/start/env.ts +++ b/start/env.ts @@ -61,6 +61,7 @@ export default await Env.create(new URL('../', import.meta.url), { LEASE_TTL_SECONDS: Env.schema.number.optional(), SPOOL_DIR: Env.schema.string.optional(), CORPUS_DIR: Env.schema.string.optional(), + CORPUS_AHEAD: Env.schema.number.optional(), PROBE_PIPELINE: Env.schema.number.optional(), SPOOL_AHEAD_PER_LEASE: Env.schema.number.optional(), SPOOL_MIN_AHEAD: Env.schema.number.optional(), diff --git a/tests/unit/services/corpus.spec.ts b/tests/unit/services/corpus.spec.ts new file mode 100644 index 0000000..8ae865e --- /dev/null +++ b/tests/unit/services/corpus.spec.ts @@ -0,0 +1,37 @@ +import { test } from '@japa/runner' +import { mkdtemp, rm } from 'node:fs/promises' +import { tmpdir } from 'node:os' +import { join } from 'node:path' +import { gzipSync } from 'node:zlib' +import coordinatorConfig from '#config/coordinator' +import Corpus from '#services/corpus' + +test.group('Corpus', (group) => { + let dir: string + let before: string | undefined + group.each.setup(async () => { + dir = await mkdtemp(join(tmpdir(), 'corpus-')) + before = coordinatorConfig.corpus.dir + coordinatorConfig.corpus.dir = dir + }) + group.each.teardown(async () => { + coordinatorConfig.corpus.dir = before + await rm(dir, { recursive: true, force: true }) + }) + + test('knows the chunks it holds, and a range once all are there', async ({ assert }) => { + const range = { group: 'alt.binaries.test', start: 1, end: 200_000 } + const first = { from: 1, to: 100_000 } + const second = { from: 100_001, to: 200_000 } + assert.isFalse(await Corpus.has(range, first)) + await Corpus.write(range, first, gzipSync('1\tsubject\n'), 2) + assert.isTrue(await Corpus.has(range, first)) + assert.isFalse(await Corpus.complete(range)) + // A reply cut short isn't kept. + await Corpus.write(range, second, Buffer.from('not gzip'), 2) + assert.isFalse(await Corpus.has(range, second)) + await Corpus.write(range, second, gzipSync('2\tsubject\n'), 2) + assert.isTrue(await Corpus.complete(range)) + assert.equal((await Corpus.read(range, second))?.length, gzipSync('2\tsubject\n').length) + }) +})