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