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 <noreply@anthropic.com>

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<boolean> {
+    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<boolean> {
+    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<Buffer | null> {
     if (!this.dir) return null

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<number>()
 
   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<number>): Promise<Target[]> {
+    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
   }
 }

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')),

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(),

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)
+  })
+})