Rebuild: hedged chunk fetches and connection max age (both off by default)

A chunk not in after NNTP_HEDGE_AFTER_SECONDS (10) is asked again on one
of NNTP_HEDGE_CONNECTIONS spare connections; the first answer wins and
the other connection is closed (lib/nntp/hedge.ts, tested). Connections
can be retired after NNTP_CONNECTION_MAX_MINUTES (±25%). The prefetch
report counts hedges and slow first requests by connection age.

Homelab, 15 min each: hedging on (40 + 6) 280k articles/s, off 273k.
The second request won about half of its races, but ~25-35% of first
requests were slow (>10 s) at every connection age, so neither helps:
Eweka is slow under our load, not on particular connections. Both
off by default; the age statistics stay.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

app/services/nntp_pool.ts

@@ -6,20 +6,38 @@ import { XoverClient } from '#lib/nntp/index'
 /**
  * The coordinator's header connections (REBUILD.md, decision e: workers
  * have no NNTP account). At most NNTP_CONNECTIONS, reused; a connection
- * that failed is closed and replaced on the next use.
+ * that failed, or is older than NNTP_CONNECTION_MAX_MINUTES, is closed and
+ * replaced on the next use.
  */
 class NntpPool {
   private idle: XoverClient[] = []
   private busy = new Set<XoverClient>()
   private open = 0
   private waiting: ((client: XoverClient) => void)[] = []
+  private openedAt = new Map<XoverClient, number>()
+  /** When a connection gets closed once free: its max age ±25%, so they don't all reconnect at once. */
+  private retireAt = new Map<XoverClient, number>()
 
   get size() {
     return coordinatorConfig.nntpConnections
   }
 
   async use<T>(work: (client: XoverClient) => Promise<T>): Promise<T> {
-    const client = await this.acquire()
+    return this.run(await this.acquire(), work)
+  }
+
+  /** As use(), if a connection is free or can be opened now; else null, without waiting. */
+  tryUse<T>(work: (client: XoverClient) => Promise<T>): Promise<T> | null {
+    if (this.idle.length === 0 && this.open >= this.size) return null
+    return this.acquire().then((client) => this.run(client, work))
+  }
+
+  /** How long ago the connection was opened, in ms. */
+  ageOf(client: XoverClient): number {
+    return Date.now() - (this.openedAt.get(client) ?? Date.now())
+  }
+
+  private async run<T>(client: XoverClient, work: (client: XoverClient) => Promise<T>): Promise<T> {
     this.busy.add(client)
     let broken = false
     try {
@@ -39,7 +57,7 @@ class NntpPool {
     if (this.open < this.size) {
       this.open++
       try {
-        return await XoverClient.connect({
+        const client = await XoverClient.connect({
           host: env.get('NNTP_HOST')!,
           port: env.get('NNTP_PORT', 563),
           user: env.get('NNTP_USER'),
@@ -48,6 +66,10 @@ class NntpPool {
           compress: true,
           timeoutMs: coordinatorConfig.nntpTimeoutMs,
         })
+        this.openedAt.set(client, Date.now())
+        const maxMs = coordinatorConfig.nntpConnectionMaxMs
+        if (maxMs > 0) this.retireAt.set(client, Date.now() + maxMs * (0.75 + Math.random() / 2))
+        return client
       } catch (error) {
         this.open--
         throw error
@@ -57,8 +79,10 @@ class NntpPool {
   }
 
   private release(client: XoverClient, broken: boolean) {
-    if (broken) {
+    if (broken || Date.now() >= (this.retireAt.get(client) ?? Infinity)) {
       this.open--
+      this.openedAt.delete(client)
+      this.retireAt.delete(client)
       try {
         client.close()
       } catch (error: any) {
@@ -112,6 +136,8 @@ class NntpPool {
     this.plain = null
     for (const client of [...this.idle.splice(0), ...this.busy]) client.close()
     this.busy.clear()
+    this.openedAt.clear()
+    this.retireAt.clear()
     this.open = 0
   }
 }

app/services/prefetcher.ts

@@ -3,6 +3,7 @@ import logger from '@adonisjs/core/services/logger'
 import coordinatorConfig from '#config/coordinator'
 import NntpPool from '#services/nntp_pool'
 import Spool, { type ChunkSpan } from '#services/spool'
+import { hedge, type Request, type XoverClient } from '#lib/nntp/index'
 
 interface Target {
   id: number
@@ -15,6 +16,9 @@ interface Target {
   total: number
 }
 
+/** Connection ages the report counts slow first requests by, in minutes. */
+const AGE_BUCKETS = [2, 10, Infinity]
+
 /** How often the targets are read again from the database. */
 const REFRESH_MS = 2000
 /** How often the spool's size is measured (a stat per file). */
@@ -27,7 +31,8 @@ const SIZE_MS = 30_000
  * order), then the next planned ranges, about SPOOL_AHEAD_PER_LEASE per
  * active lease (at least SPOOL_MIN_AHEAD), within SPOOL_MAX_GB. One
  * connection fills ~26 k overview lines/s, so a range on one connection
- * kept its worker waiting (LOG.md 2026-10-09).
+ * kept its worker waiting (LOG.md 2026-10-09). A chunk that takes long is
+ * asked again on a spare connection (fetchHedged).
  */
 export class Prefetcher {
   /**
@@ -45,7 +50,25 @@ export class Prefetcher {
   private spoolBytes = 0
   private sizedAt = 0
   /** Fetched since the last report: chunks, articles, ms spent fetching. */
-  private stats = { chunks: 0, articles: 0, fetchMs: 0, since: Date.now() }
+  private stats = this.freshStats()
+
+  private freshStats() {
+    return {
+      chunks: 0,
+      articles: 0,
+      fetchMs: 0,
+      hedged: 0,
+      hedgeWon: 0,
+      /** First requests per connection-age bucket, and how many of them were slow. */
+      byAge: AGE_BUCKETS.map(() => ({ requests: 0, slow: 0 })),
+      since: Date.now(),
+    }
+  }
+
+  /** The prefetcher's loops: the pool less the connections kept for hedging. */
+  private get loops() {
+    return Math.max(1, NntpPool.size - coordinatorConfig.nntpHedgeConnections)
+  }
 
   /** One round: refreshes the targets now and then, and puts idle connections to work. */
   async tick() {
@@ -54,7 +77,7 @@ export class Prefetcher {
       this.targets = await this.readTargets()
     }
     this.report()
-    while (!this.stopped && this.running < NntpPool.size && this.peek()) {
+    while (!this.stopped && this.running < this.loops && this.peek()) {
       this.running++
       this.work()
         .catch((error) => logger.warn(`Prefetching: ${error.message}`))
@@ -82,7 +105,7 @@ export class Prefetcher {
           await client.group(target.group)
           return client.xoverRaw(span.from, span.to)
         }
-        const data = target.plain.has(span.from) ? await NntpPool.usePlain(fetch) : await NntpPool.use(fetch)
+        const data = target.plain.has(span.from) ? await NntpPool.usePlain(fetch) : await this.fetchHedged(target.group, span)
         this.stats.chunks++
         this.stats.articles += span.to - span.from + 1
         this.stats.fetchMs += Date.now() - fetchedAt
@@ -101,18 +124,69 @@ export class Prefetcher {
     }
   }
 
+  /**
+   * A chunk over a pooled connection. Not in after nntpHedgeAfterMs, it's
+   * asked again on a free connection (if there is one): Eweka stalls single
+   * requests, and the same chunk on another connection mostly comes in
+   * seconds (LOG.md 2026-10-09 (15)). The first answer wins; the other
+   * request can't be cancelled, so its connection is closed (the pool opens
+   * a fresh one).
+   */
+  private async fetchHedged(group: string, span: ChunkSpan): Promise<Buffer> {
+    let age: number | undefined
+    const request = (start: (work: (client: XoverClient) => Promise<Buffer>) => Promise<Buffer> | null, first: boolean): Request<Buffer> | null => {
+      let client: XoverClient | undefined
+      let cancelled = false
+      const done = start(async (c) => {
+        // Given up before it got a connection: the connection goes back unused.
+        if (cancelled) return Buffer.alloc(0)
+        client = c
+        if (first) age = NntpPool.ageOf(c)
+        await c.group(group)
+        return c.xoverRaw(span.from, span.to)
+      })
+      if (!done) return null
+      done.catch(() => {})
+      return {
+        done,
+        // The request can't be called off on the wire: its connection is closed, and the pool opens a fresh one.
+        cancel: () => {
+          cancelled = true
+          client?.close()
+        },
+      }
+    }
+    const result = await hedge(
+      () => request((work) => NntpPool.use(work), true)!,
+      () => request((work) => NntpPool.tryUse(work), false),
+      coordinatorConfig.nntpHedgeAfterMs
+    )
+    if (age !== undefined) {
+      const bucket = this.stats.byAge[AGE_BUCKETS.findIndex((minutes) => age! < minutes * 60_000)]
+      bucket.requests++
+      if (result.late) bucket.slow++
+    }
+    if (result.hedged) this.stats.hedged++
+    if (result.secondWon) this.stats.hedgeWon++
+    return result.value
+  }
+
   /** A line a minute: what the connections fetched, and how long a chunk took. */
   private report() {
     const seconds = (Date.now() - this.stats.since) / 1000
     if (seconds < 60) return
-    const { chunks, articles, fetchMs } = this.stats
+    const { chunks, articles, fetchMs, hedged, hedgeWon, byAge } = this.stats
     if (chunks > 0) {
+      const ages = byAge
+        .map((b, i) => `${i === 0 ? '<' : `${AGE_BUCKETS[i - 1]}–`}${AGE_BUCKETS[i] === Infinity ? '' : AGE_BUCKETS[i]} min ${b.slow}/${b.requests}`.replace('– min', '+ min'))
+        .join(', ')
       logger.info(
         `Prefetched ${chunks} chunks in ${seconds.toFixed(0)} s: ${Math.round(articles / seconds).toLocaleString('en')} articles/s, ` +
-          `${(fetchMs / chunks / 1000).toFixed(1)} s a chunk, connections busy ${Math.round((fetchMs / (seconds * 1000 * NntpPool.size)) * 100)}%.`
+          `${(fetchMs / chunks / 1000).toFixed(1)} s a chunk, loops busy ${Math.round((fetchMs / (seconds * 1000 * this.loops)) * 100)}%; ` +
+          `hedged ${hedged} (second won ${hedgeWon}); slow by connection age: ${ages}.`
       )
     }
-    this.stats = { chunks: 0, articles: 0, fetchMs: 0, since: Date.now() }
+    this.stats = this.freshStats()
   }
 
   private peek(): boolean {

config/coordinator.ts

@@ -37,6 +37,24 @@ export default {
    * the prefetcher nearly stopped. Data arrives within seconds otherwise.
    */
   nntpTimeoutMs: env.get('NNTP_TIMEOUT_MS', 2 * 60 * 1000),
+  /**
+   * A chunk not fetched after this long is asked again on another
+   * connection; the first answer wins and the other connection is closed.
+   * Eweka stalls single requests: the same chunk took 27–52 s on one
+   * connection and 1–9 s on another at the same moment; median ~3 s
+   * (LOG.md 2026-10-09 (15)). Measured in production (40 + 6 connections vs
+   * 40, 15 min each): 280k vs 273k articles/s, within the noise; the second
+   * request won half the time. Off by default.
+   */
+  nntpHedgeAfterMs: env.get('NNTP_HEDGE_AFTER_SECONDS', 10) * 1000,
+  /** Of nntpConnections, kept for those second requests (the prefetcher's loops are the rest). 0: no hedging. */
+  nntpHedgeConnections: env.get('NNTP_HEDGE_CONNECTIONS', 0),
+  /**
+   * A connection is closed after this long (when it's free), and a fresh one
+   * opened. 0: never. Requests were as often slow (>10 s) on connections
+   * under 2 minutes old as on ones over 10 (~25–35%; LOG.md 2026-10-09 (16)).
+   */
+  nntpConnectionMaxMs: env.get('NNTP_CONNECTION_MAX_MINUTES', 0) * 60 * 1000,
   spool: {
     dir: env.get('SPOOL_DIR', app.tmpPath('spool')),
     /** Ranges kept fetched ahead, per active lease. */

lib/nntp/hedge.ts

@@ -0,0 +1,51 @@
+/** A request that can be given up: cancel() stops it (closing its connection, say). */
+export interface Request<T> {
+  done: Promise<T>
+  cancel(): void
+}
+
+export interface Hedged<T> {
+  value: T
+  /** The first request took longer than afterMs. */
+  late: boolean
+  /** A second request was started. */
+  hedged: boolean
+  /** The second request answered first. */
+  secondWon: boolean
+}
+
+/**
+ * A hedged request: the first one not answered within afterMs is asked again
+ * (again() may decline with null: no connection free), and the first answer
+ * wins; the other request is cancelled. A failure only counts once both failed.
+ */
+export async function hedge<T>(start: () => Request<T>, again: () => Request<T> | null, afterMs: number): Promise<Hedged<T>> {
+  const first = start()
+  let timer: NodeJS.Timeout | undefined
+  const late = await Promise.race([
+    first.done.then(
+      () => false,
+      () => false
+    ),
+    new Promise<boolean>((resolve) => (timer = setTimeout(() => resolve(true), afterMs))),
+  ])
+  clearTimeout(timer)
+  if (!late) return { value: await first.done, late, hedged: false, secondWon: false }
+  const second = again()
+  if (!second) return { value: await first.done, late, hedged: false, secondWon: false }
+  const settled = new Set<Request<T>>()
+  for (const r of [first, second]) {
+    r.done.then(
+      () => settled.add(r),
+      () => settled.add(r)
+    )
+  }
+  try {
+    const winner = await Promise.any([first, second].map((r) => r.done.then((value) => ({ r, value }))))
+    const loser = winner.r === first ? second : first
+    if (!settled.has(loser)) loser.cancel()
+    return { value: winner.value, late, hedged: true, secondWon: winner.r === second }
+  } catch (error) {
+    throw error instanceof AggregateError ? error.errors[0] : error
+  }
+}

lib/nntp/index.ts

@@ -1,3 +1,4 @@
 export { XoverClient, parseOverviews, type Overview, type XoverClientOptions } from './xover_client.js'
 export { parseXrefGroups } from './xref.js'
 export { parseGroupResponse, type GroupInfo } from './group.js'
+export { hedge, type Request, type Hedged } from './hedge.js'

start/env.ts

@@ -44,6 +44,9 @@ export default await Env.create(new URL('../', import.meta.url), {
   NNTP_SECURE: Env.schema.boolean.optional(),
   NNTP_CONNECTIONS: Env.schema.number.optional(),
   NNTP_TIMEOUT_MS: Env.schema.number.optional(),
+  NNTP_HEDGE_AFTER_SECONDS: Env.schema.number.optional(),
+  NNTP_HEDGE_CONNECTIONS: Env.schema.number.optional(),
+  NNTP_CONNECTION_MAX_MINUTES: Env.schema.number.optional(),
 
   // Newznab API key (empty: no key needed)
   NEWZNAB_API_KEY: Env.schema.secret.optional(),

tests/unit/nntp/hedge.spec.ts

@@ -0,0 +1,50 @@
+import { test } from '@japa/runner'
+import { hedge, type Request } from '#lib/nntp/index'
+
+/** A request answering `value` after ms (or failing), recording whether it was cancelled. */
+function request<T>(ms: number, value: T, fail = false) {
+  const r = { cancelled: false } as Request<T> & { cancelled: boolean }
+  r.done = new Promise<T>((resolve, reject) => setTimeout(() => (fail ? reject(new Error(String(value))) : resolve(value)), ms))
+  r.cancel = () => (r.cancelled = true)
+  return r
+}
+
+test.group('nntp / hedge', () => {
+  test('a quick answer: no second request', async ({ assert }) => {
+    let asked = 0
+    const result = await hedge(() => request(5, 'a'), () => (asked++, request(5, 'b')), 50)
+    assert.deepEqual(result, { value: 'a', late: false, hedged: false, secondWon: false })
+    assert.equal(asked, 0)
+  })
+
+  test('a stalled request: the second answers, the first is cancelled', async ({ assert }) => {
+    const first = request(1000, 'a')
+    const result = await hedge(() => first, () => request(10, 'b'), 20)
+    assert.deepEqual(result, { value: 'b', late: true, hedged: true, secondWon: true })
+    assert.isTrue(first.cancelled)
+  })
+
+  test('late, but the first still wins: the second is cancelled', async ({ assert }) => {
+    const second = request(1000, 'b')
+    const result = await hedge(() => request(40, 'a'), () => second, 20)
+    assert.deepEqual(result, { value: 'a', late: true, hedged: true, secondWon: false })
+    assert.isTrue(second.cancelled)
+  })
+
+  test('no connection free: waits for the first', async ({ assert }) => {
+    const result = await hedge(() => request(40, 'a'), () => null, 10)
+    assert.deepEqual(result, { value: 'a', late: true, hedged: false, secondWon: false })
+  })
+
+  test('one fails: the other answers; both fail: the first error', async ({ assert }) => {
+    const result = await hedge(() => request(30, 'a', true), () => request(40, 'b'), 10)
+    assert.equal(result.value, 'b')
+    await assert.rejects(() => hedge(() => request(30, 'a', true), () => request(40, 'b', true), 10), 'a')
+  })
+
+  test('a failure before afterMs fails at once', async ({ assert }) => {
+    let asked = 0
+    await assert.rejects(() => hedge(() => request(5, 'a', true), () => (asked++, request(5, 'b')), 50), 'a')
+    assert.equal(asked, 0)
+  })
+})