From 40aa78cda39a68b31dcda098ee038bbbd3d30f1a Mon Sep 17 00:00:00 2001 From: Daan Meijer Date: Sat, 10 Oct 2026 08:30:04 +0200 Subject: [PATCH] Rebuild: a deadline per chunk; batches with a huge fragment A worker waited on one chunk of a leased range for an hour (range 146477: five 15-minute stream timeouts, then the range took 63 s; 88 "stream: Peer disconnected" overnight). The silence timeout doesn't catch a reply that trickles: a chunk still coming in after NNTP_CHUNK_DEADLINE_SECONDS (300) is given up, its connection closed, and fetched again. The report names the oldest request in flight (range, chunk, age, bytes in) and counts deadlines. Batches failed with "Maximum call stack size exceeded" (8 overnight): Math.min(...segments) on a fragment with more segments than the stack takes arguments. Co-Authored-By: Claude Opus 5.5 --- app/services/batch_store.ts | 3 ++- app/services/prefetcher.ts | 45 +++++++++++++++++++++++++++++++------ config/coordinator.ts | 7 ++++++ lib/nntp/xover_client.ts | 4 ++++ start/env.ts | 1 + 5 files changed, 52 insertions(+), 8 deletions(-) diff --git a/app/services/batch_store.ts b/app/services/batch_store.ts index 14c9673..11a5b6b 100644 --- a/app/services/batch_store.ts +++ b/app/services/batch_store.ts @@ -77,7 +77,8 @@ class BatchStore { const rows = batch.fragments.slice(i, i + 2000).map((f) => ({ range_id: rangeId, file_key: f.fileKey, - article_number: Math.min(...f.segments.map((s) => s[3])), + // Not Math.min(...): a big file's fragment has more segments than the call stack takes arguments. + article_number: f.segments.reduce((min, s) => (s[3] < min ? s[3] : min), Infinity), segments: f.segments.length, fragment: JSON.stringify(f), })) diff --git a/app/services/prefetcher.ts b/app/services/prefetcher.ts index 13a10be..8048873 100644 --- a/app/services/prefetcher.ts +++ b/app/services/prefetcher.ts @@ -59,6 +59,8 @@ export class Prefetcher { private fetchingNow = 0 private writingNow = 0 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 }>() private freshStats() { return { @@ -84,6 +86,8 @@ export class Prefetcher { idle: 0, inFlight: 0, inFlightMin: Infinity, + /** Requests given up at nntpChunkDeadlineMs. */ + deadlines: 0, ranges: 0, groups: 0, since: Date.now(), @@ -135,7 +139,7 @@ export class Prefetcher { } let data: Buffer try { - data = target.plain.has(span.from) ? await NntpPool.usePlain(fetch) : await this.fetchHedged(target.group, span) + data = target.plain.has(span.from) ? await NntpPool.usePlain(fetch) : await this.fetchHedged(target.id, target.group, span) } finally { this.fetchingNow-- } @@ -173,7 +177,7 @@ export class Prefetcher { * request can't be cancelled, so its connection is closed (the pool opens * a fresh one). */ - private async fetchHedged(group: string, span: ChunkSpan): Promise { + private async fetchHedged(rangeId: number, 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 @@ -183,10 +187,28 @@ export class Prefetcher { if (cancelled) return Buffer.alloc(0) client = c if (first) age = NntpPool.ageOf(c) - const groupAt = performance.now() - await c.group(group) - const groupMs = performance.now() - groupAt - const data = await c.xoverRaw(span.from, span.to) + const entry = { rangeId, from: span.from, to: span.to, since: Date.now(), client: c } + this.inFlight.add(entry) + let late = false + const deadline = setTimeout(() => { + late = true + this.stats.deadlines++ + c.close() + }, coordinatorConfig.nntpChunkDeadlineMs) + let data: Buffer + let groupMs = 0 + try { + const groupAt = performance.now() + await c.group(group) + groupMs = performance.now() - groupAt + data = await c.xoverRaw(span.from, span.to) + } catch (error) { + if (!late) throw error + throw new Error(`Chunk ${span.from}–${span.to} of range ${rangeId} not in after ${coordinatorConfig.nntpChunkDeadlineMs / 1000} s; fetching it again`) + } finally { + clearTimeout(deadline) + this.inFlight.delete(entry) + } if (!cancelled) { const t = c.lastTiming this.stats.requests++ @@ -223,6 +245,14 @@ export class Prefetcher { return result.value } + /** The longest-running request: range, chunk, how long, how much came in. */ + private oldest() { + let oldest: { rangeId: number; from: number; to: number; since: number; client: XoverClient } | undefined + for (const r of this.inFlight) if (!oldest || r.since < oldest.since) oldest = r + if (!oldest) return 'none in flight' + return `oldest request ${Math.round((Date.now() - oldest.since) / 1000)} s (range ${oldest.rangeId}, ${oldest.from}–${oldest.to}, ${Math.round(oldest.client.buffered / 1000)} kB in)` + } + private sample() { const s = this.stats s.samples++ @@ -257,7 +287,8 @@ export class Prefetcher { `on ${avg(st.ranges).toFixed(1)} ranges of ${avg(st.groups).toFixed(1)} groups; ` + `loops ${avg(st.fetching).toFixed(1)} fetching, ${avg(st.writing).toFixed(1)} writing, ${avg(st.idle).toFixed(1)} idle; ` + `a request: GROUP ${per(st.groupMs)} s, XOVER reply ${per(st.replyMs)} s, transfer ${per(st.bodyMs)} s; ` + - `${st.silent} of ${requests} transfers silent >5 s; writing ${(st.writeMs / Math.max(1, chunks)).toFixed(0)} ms a chunk.` + `${st.silent} of ${requests} transfers silent >5 s; writing ${(st.writeMs / Math.max(1, chunks)).toFixed(0)} ms a chunk; ` + + `${st.deadlines} given up after ${coordinatorConfig.nntpChunkDeadlineMs / 1000} s; ${this.oldest()}.` ) } this.stats = this.freshStats() diff --git a/config/coordinator.ts b/config/coordinator.ts index d9acb4b..2be09ed 100644 --- a/config/coordinator.ts +++ b/config/coordinator.ts @@ -37,6 +37,13 @@ export default { * the prefetcher nearly stopped. Data arrives within seconds otherwise. */ nntpTimeoutMs: env.get('NNTP_TIMEOUT_MS', 2 * 60 * 1000), + /** + * A chunk still coming in after this long is given up (its connection + * closed) and fetched again: a reply that trickles never goes quiet long + * enough for nntpTimeoutMs, and kept a worker waiting on one chunk for an + * hour (LOG.md 2026-10-10 (2)). Chunks take 3–35 s. + */ + nntpChunkDeadlineMs: env.get('NNTP_CHUNK_DEADLINE_SECONDS', 5 * 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. diff --git a/lib/nntp/xover_client.ts b/lib/nntp/xover_client.ts index c985aac..846e6e0 100644 --- a/lib/nntp/xover_client.ts +++ b/lib/nntp/xover_client.ts @@ -47,6 +47,10 @@ export class XoverClient { */ lastTiming = { replyMs: 0, bodyMs: 0, longestGapMs: 0 } private measuring = false + /** Bytes received and not consumed yet: during a reply, how much of it came in. */ + get buffered() { + return this.length + } private lastDataAt = 0 private longestGap = 0 diff --git a/start/env.ts b/start/env.ts index be506a2..671e2d2 100644 --- a/start/env.ts +++ b/start/env.ts @@ -47,6 +47,7 @@ export default await Env.create(new URL('../', import.meta.url), { NNTP_HEDGE_AFTER_SECONDS: Env.schema.number.optional(), NNTP_HEDGE_CONNECTIONS: Env.schema.number.optional(), NNTP_CONNECTION_MAX_MINUTES: Env.schema.number.optional(), + NNTP_CHUNK_DEADLINE_SECONDS: Env.schema.number.optional(), // Newznab API key (empty: no key needed) NEWZNAB_API_KEY: Env.schema.secret.optional(),