From 84759691458de46183ca9a60c0722820ee9ff815 Mon Sep 17 00:00:00 2001 From: Daan Meijer Date: Fri, 9 Oct 2026 22:13:44 +0200 Subject: [PATCH] Rebuild: hedged chunk fetches and connection max age (both off by default) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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 --- app/services/nntp_pool.ts | 34 ++++++++++++-- app/services/prefetcher.ts | 88 ++++++++++++++++++++++++++++++++--- config/coordinator.ts | 18 +++++++ lib/nntp/hedge.ts | 51 ++++++++++++++++++++ lib/nntp/index.ts | 1 + start/env.ts | 3 ++ tests/unit/nntp/hedge.spec.ts | 50 ++++++++++++++++++++ 7 files changed, 234 insertions(+), 11 deletions(-) create mode 100644 lib/nntp/hedge.ts create mode 100644 tests/unit/nntp/hedge.spec.ts diff --git a/app/services/nntp_pool.ts b/app/services/nntp_pool.ts index 622076b..faebf22 100644 --- a/app/services/nntp_pool.ts +++ b/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() private open = 0 private waiting: ((client: XoverClient) => void)[] = [] + private openedAt = new Map() + /** When a connection gets closed once free: its max age ±25%, so they don't all reconnect at once. */ + private retireAt = new Map() get size() { return coordinatorConfig.nntpConnections } async use(work: (client: XoverClient) => Promise): Promise { - 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(work: (client: XoverClient) => Promise): Promise | 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(client: XoverClient, work: (client: XoverClient) => Promise): Promise { 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 } } diff --git a/app/services/prefetcher.ts b/app/services/prefetcher.ts index e828078..01324a9 100644 --- a/app/services/prefetcher.ts +++ b/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 { + let age: number | undefined + const request = (start: (work: (client: XoverClient) => Promise) => Promise | null, first: boolean): Request | 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 { diff --git a/config/coordinator.ts b/config/coordinator.ts index 7faa61a..7266571 100644 --- a/config/coordinator.ts +++ b/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. */ diff --git a/lib/nntp/hedge.ts b/lib/nntp/hedge.ts new file mode 100644 index 0000000..6c1e3a2 --- /dev/null +++ b/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 { + done: Promise + cancel(): void +} + +export interface Hedged { + 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(start: () => Request, again: () => Request | null, afterMs: number): Promise> { + const first = start() + let timer: NodeJS.Timeout | undefined + const late = await Promise.race([ + first.done.then( + () => false, + () => false + ), + new Promise((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>() + 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 + } +} diff --git a/lib/nntp/index.ts b/lib/nntp/index.ts index c80c36b..6dea896 100644 --- a/lib/nntp/index.ts +++ b/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' diff --git a/start/env.ts b/start/env.ts index 165d92d..4fb72d2 100644 --- a/start/env.ts +++ b/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(), diff --git a/tests/unit/nntp/hedge.spec.ts b/tests/unit/nntp/hedge.spec.ts new file mode 100644 index 0000000..7753a66 --- /dev/null +++ b/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(ms: number, value: T, fail = false) { + const r = { cancelled: false } as Request & { cancelled: boolean } + r.done = new Promise((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) + }) +})