Daan MeijerandClaude Opus 5.5
40aa78cda3
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 <noreply@anthropic.com>
361 lines
15 KiB
TypeScript
361 lines
15 KiB
TypeScript
import db from '@adonisjs/lucid/services/db'
|
||
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
|
||
group: string
|
||
/** Chunk starts to fetch without compression (ranges.plain_chunks). */
|
||
plain: Set<number>
|
||
/** Chunks not spooled yet, in order. */
|
||
missing: ChunkSpan[]
|
||
/** All of the range's chunks, from its start. */
|
||
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). */
|
||
const SIZE_MS = 30_000
|
||
|
||
/**
|
||
* Keeps the news server busy ahead of the workers (REBUILD.md, prefetch).
|
||
* Every connection takes the next missing chunk: the leased ranges' first
|
||
* (a worker waits for them), then the next planned ranges, with at most
|
||
* PREFETCH_PER_RANGE connections on a range and PREFETCH_PER_GROUP on a
|
||
* group (Eweka answers faster spread out), 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). A chunk that takes long is
|
||
* asked again on a spare connection (fetchHedged).
|
||
*/
|
||
export class Prefetcher {
|
||
/**
|
||
* This process's share of the ranges, `id % count = index` (`coordinator:run --shard`): one
|
||
* process's event loop handles ~60 MB/s of compressed XOVER (16.5% of a core at ~10 MB/s;
|
||
* LOG.md 2026-10-09), less than 50 connections bring.
|
||
*/
|
||
constructor(private shard = { index: 0, count: 1 }) {}
|
||
|
||
private targets: Target[] = []
|
||
private claimed = new Set<string>()
|
||
/** Chunks being fetched per range and per group (spool.perRange, spool.perGroup). */
|
||
private perRange = new Map<number, number>()
|
||
private perGroup = new Map<string, number>()
|
||
private running = 0
|
||
private refreshedAt = 0
|
||
private stopped = false
|
||
private spoolBytes = 0
|
||
private sizedAt = 0
|
||
/** Fetched since the last report: chunks, articles, ms spent fetching. */
|
||
private stats = this.freshStats()
|
||
/** Loops in a fetch or a spool write right now. */
|
||
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 {
|
||
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 })),
|
||
/** Spool writes and the range's update, ms. */
|
||
writeMs: 0,
|
||
/** Per answered request, ms: GROUP, XOVER until its status line, the transfer; transfers with a silence over 5 s. */
|
||
requests: 0,
|
||
groupMs: 0,
|
||
replyMs: 0,
|
||
bodyMs: 0,
|
||
silent: 0,
|
||
/** Once a second: loops fetching, writing, not running; pool connections with a command in flight. */
|
||
samples: 0,
|
||
fetching: 0,
|
||
writing: 0,
|
||
idle: 0,
|
||
inFlight: 0,
|
||
inFlightMin: Infinity,
|
||
/** Requests given up at nntpChunkDeadlineMs. */
|
||
deadlines: 0,
|
||
ranges: 0,
|
||
groups: 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() {
|
||
this.sampler ??= setInterval(() => this.sample(), 1000).unref()
|
||
if (Date.now() - this.refreshedAt >= REFRESH_MS) {
|
||
this.refreshedAt = Date.now()
|
||
this.targets = await this.readTargets()
|
||
}
|
||
this.report()
|
||
while (!this.stopped && this.running < this.loops && this.peek()) {
|
||
this.running++
|
||
this.work()
|
||
.catch((error) => logger.warn(`Prefetching: ${error.message}`))
|
||
.finally(() => this.running--)
|
||
}
|
||
}
|
||
|
||
stop() {
|
||
this.stopped = true
|
||
clearInterval(this.sampler)
|
||
}
|
||
|
||
get active() {
|
||
return this.running
|
||
}
|
||
|
||
/** One connection's loop: chunks until there is nothing left to fetch. */
|
||
private async work() {
|
||
while (!this.stopped) {
|
||
const next = this.take()
|
||
if (!next) return
|
||
const { target, span } = next
|
||
try {
|
||
const fetchedAt = Date.now()
|
||
this.fetchingNow++
|
||
const fetch = async (client: Parameters<Parameters<typeof NntpPool.use>[0]>[0]) => {
|
||
await client.group(target.group)
|
||
return client.xoverRaw(span.from, span.to)
|
||
}
|
||
let data: Buffer
|
||
try {
|
||
data = target.plain.has(span.from) ? await NntpPool.usePlain(fetch) : await this.fetchHedged(target.id, target.group, span)
|
||
} finally {
|
||
this.fetchingNow--
|
||
}
|
||
this.stats.chunks++
|
||
this.stats.articles += span.to - span.from + 1
|
||
this.stats.fetchMs += Date.now() - fetchedAt
|
||
const writtenAt = Date.now()
|
||
this.writingNow++
|
||
try {
|
||
const existed = await Spool.has(target.id, span)
|
||
await Spool.write(target.id, span, data)
|
||
if (existed) continue
|
||
this.spoolBytes += data.length
|
||
const { rows } = await db.rawQuery(
|
||
`UPDATE ranges SET spooled_chunks = spooled_chunks + 1 WHERE id = ? RETURNING spooled_chunks`,
|
||
[target.id]
|
||
)
|
||
if (Number(rows[0]?.spooled_chunks) >= target.total) await db.from('ranges').where('id', target.id).update({ spooled_at: new Date() })
|
||
} finally {
|
||
this.writingNow--
|
||
this.stats.writeMs += Date.now() - writtenAt
|
||
}
|
||
} finally {
|
||
this.claimed.delete(`${target.id}:${span.from}`)
|
||
this.count(target, -1)
|
||
}
|
||
}
|
||
}
|
||
|
||
/**
|
||
* 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(rangeId: number, 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)
|
||
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++
|
||
this.stats.groupMs += groupMs
|
||
this.stats.replyMs += t.replyMs
|
||
this.stats.bodyMs += t.bodyMs
|
||
if (t.longestGapMs > 5000) this.stats.silent++
|
||
}
|
||
return data
|
||
})
|
||
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
|
||
}
|
||
|
||
/** 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++
|
||
s.fetching += this.fetchingNow
|
||
s.writing += this.writingNow
|
||
s.idle += Math.max(0, this.loops - this.fetchingNow - this.writingNow)
|
||
s.inFlight += NntpPool.busyCount
|
||
s.inFlightMin = Math.min(s.inFlightMin, NntpPool.busyCount)
|
||
s.ranges += this.perRange.size
|
||
s.groups += this.perGroup.size
|
||
}
|
||
|
||
/** 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, hedged, hedgeWon, byAge, samples, requests } = 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; ` +
|
||
`hedged ${hedged} (second won ${hedgeWon}); slow by connection age: ${ages}.`
|
||
)
|
||
const st = this.stats
|
||
const avg = (n: number) => (samples > 0 ? n / samples : 0)
|
||
const per = (ms: number) => (requests > 0 ? (ms / requests / 1000).toFixed(1) : '-')
|
||
logger.info(
|
||
`Connections: ${avg(st.inFlight).toFixed(1)} of ${NntpPool.size} with a command in flight (min ${st.inFlightMin}), ` +
|
||
`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.deadlines} given up after ${coordinatorConfig.nntpChunkDeadlineMs / 1000} s; ${this.oldest()}.`
|
||
)
|
||
}
|
||
this.stats = this.freshStats()
|
||
}
|
||
|
||
private peek(): boolean {
|
||
return this.targets.some((t) => this.open(t) && t.missing.some((s) => !this.claimed.has(`${t.id}:${s.from}`)))
|
||
}
|
||
|
||
/** Whether the target is under its range's and its group's limit of connections. */
|
||
private open(target: Target) {
|
||
const { perRange, perGroup } = coordinatorConfig.spool
|
||
return (perRange <= 0 || (this.perRange.get(target.id) ?? 0) < perRange) && (perGroup <= 0 || (this.perGroup.get(target.group) ?? 0) < perGroup)
|
||
}
|
||
|
||
private count(target: Target, by: number) {
|
||
const range = (this.perRange.get(target.id) ?? 0) + by
|
||
const group = (this.perGroup.get(target.group) ?? 0) + by
|
||
if (range > 0) this.perRange.set(target.id, range)
|
||
else this.perRange.delete(target.id)
|
||
if (group > 0) this.perGroup.set(target.group, group)
|
||
else this.perGroup.delete(target.group)
|
||
}
|
||
|
||
/** The first unclaimed missing chunk, in target order, of a target under its limits. */
|
||
private take(): { target: Target; span: ChunkSpan } | null {
|
||
for (const target of this.targets) {
|
||
if (!this.open(target)) continue
|
||
const index = target.missing.findIndex((s) => !this.claimed.has(`${target.id}:${s.from}`))
|
||
if (index === -1) continue
|
||
const [span] = target.missing.splice(index, 1)
|
||
this.claimed.add(`${target.id}:${span.from}`)
|
||
this.count(target, 1)
|
||
return { target, span }
|
||
}
|
||
return null
|
||
}
|
||
|
||
private async readTargets(): Promise<Target[]> {
|
||
if (Date.now() - this.sizedAt >= SIZE_MS) {
|
||
this.sizedAt = Date.now()
|
||
this.spoolBytes = await Spool.size()
|
||
}
|
||
const mine = (q: any) => (this.shard.count > 1 ? q.whereRaw('r.id % ? = ?', [this.shard.count, this.shard.index]) : q)
|
||
const rows = (q: any) => mine(q.from('ranges as r').join('groups as g', 'g.id', 'r.group_id').select('r.id', 'g.name as group', 'r.start_id', 'r.end_id', 'r.next_id', 'r.plain_chunks'))
|
||
// Leased ranges, checked against the spool's files whatever spooled_at says: a worker waits for them.
|
||
const leased = await rows(db.query()).where('r.status', 'leased').orderBy('r.leased_at')
|
||
// Planned ranges ahead: those started already, then new ones up to SPOOL_AHEAD_PER_LEASE per lease.
|
||
const ahead = Math.max(coordinatorConfig.spool.minAhead, coordinatorConfig.spool.aheadPerLease * leased.length)
|
||
const started = await rows(db.query()).where('r.status', 'planned').where('r.spooled_chunks', '>', 0).whereNull('r.spooled_at').orderBy([{ column: 'r.priority' }, { column: 'r.id' }])
|
||
const [{ ready }] = await mine(db.from('ranges as r').where('r.status', 'planned').where('r.spooled_chunks', '>', 0)).count('* as ready')
|
||
const room = this.spoolBytes < coordinatorConfig.spool.maxBytes
|
||
const fresh = room && Number(ready) < ahead
|
||
? await rows(db.query()).where('r.status', 'planned').where('r.spooled_chunks', 0).whereRaw('r.next_id <= r.end_id').orderBy([{ column: 'r.priority' }, { column: 'r.id' }]).limit(ahead - Number(ready))
|
||
: []
|
||
const targets: Target[] = []
|
||
for (const row of [...leased, ...(room ? started : []), ...fresh]) {
|
||
const start = Number(row.start_id)
|
||
const end = Number(row.end_id)
|
||
const total = Spool.chunksOf({ start, end }).length
|
||
const missing: ChunkSpan[] = []
|
||
for (const span of Spool.chunksOf({ start, end }, Number(row.next_id))) {
|
||
if (!this.claimed.has(`${row.id}:${span.from}`) && !(await Spool.has(Number(row.id), span))) missing.push(span)
|
||
}
|
||
if (missing.length > 0) targets.push({ id: Number(row.id), group: row.group, plain: new Set((row.plain_chunks ?? []).map(Number)), missing, total })
|
||
}
|
||
return targets
|
||
}
|
||
}
|