Daan MeijerandClaude Opus 5.5
b60d716db0
Eweka drops quiet connections, so the next command on one failed
("Spot inspection failed: NNTP connection closed"), and idle connections
held account slots that downloads could use. The coordinator pool now
closes free connections (and the plain one) on its sync once idle for
60 s; dry prober lanes, the spotter between rounds and the spot inspector
with nothing to inspect close theirs too, reconnecting when work comes.
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
370 lines
18 KiB
TypeScript
370 lines
18 KiB
TypeScript
import db from '@adonisjs/lucid/services/db'
|
||
import logger from '@adonisjs/core/services/logger'
|
||
import { setTimeout as sleep } from 'node:timers/promises'
|
||
import env from '#start/env'
|
||
import coordinatorConfig from '#config/coordinator'
|
||
import { XoverClient } from '#lib/nntp/index'
|
||
import { YencError } from '#lib/yenc/index'
|
||
import { inspect, innerName, PROBE_VERSION, type Inspection } from '#lib/probe/index'
|
||
import { NntpShare } from '#services/nntp_shares'
|
||
import { DiskSpace } from '#services/disk_space'
|
||
|
||
interface Candidate {
|
||
id: number
|
||
releaseId: number
|
||
kind: string
|
||
messageId: string
|
||
}
|
||
|
||
type Outcome = { candidate: Candidate; status: 'done' | 'missing' | 'error' | 'waiting'; inspection?: Inspection; cached?: boolean }
|
||
|
||
/**
|
||
* The order candidates are probed in: what the par2 merge needs first (a random-named file's own start, then the
|
||
* par2 sets and archives of releases without a readable name), then the rest (names and passwords of readable ones).
|
||
*/
|
||
const TIERS: { kind: string; unreadable: boolean }[] = [
|
||
{ kind: 'lone', unreadable: false },
|
||
{ kind: 'par2', unreadable: true },
|
||
{ kind: 'archive', unreadable: true },
|
||
{ kind: 'par2', unreadable: false },
|
||
{ kind: 'archive', unreadable: false },
|
||
]
|
||
const FLUSH_MS = 2000
|
||
|
||
/**
|
||
* JSON for a jsonb parameter, without NUL characters: jsonb can't hold \u0000 (a name read from an article had one),
|
||
* and the whole flush failed with it (LOG.md 2026-10-18 (78)).
|
||
*/
|
||
function jsonb(value: unknown): string {
|
||
return JSON.stringify(value, (_key, v) => (typeof v === 'string' ? v.replaceAll('\0', '') : v))
|
||
}
|
||
|
||
/** Slices a claim takes from random points of the queue. */
|
||
const SLICES = 20
|
||
const REPORT_MS = 60_000
|
||
|
||
/**
|
||
* Phase 2's first step (REBUILD.md, part 6): reads the probe queue's
|
||
* articles (settled releases first, by kind) over PROBE_CONNECTIONS
|
||
* connections of its own, without compression, and stores what they show:
|
||
* per release and kind an Inspection, and on the release whether it's
|
||
* passworded and its inner name. One process: fetching (~0.1 s an article)
|
||
* dwarfs parsing (a few ms), so there's no queue between them.
|
||
*/
|
||
export class Prober {
|
||
private queue: Candidate[] = []
|
||
private outcomes: Outcome[] = []
|
||
private refilling: Promise<void> | null = null
|
||
private stopped = false
|
||
private stats = Prober.freshStats()
|
||
/** Lanes running (each one connection, once it has work); at most the share's grant. Counted apart from the set: a lane leaving takes itself off at once. */
|
||
private lanes = new Set<Promise<void>>()
|
||
private running = 0
|
||
/** The last lane to look found nothing to claim: the prober wants its minimum only. */
|
||
private dry = false
|
||
/** What it stores (inspections, par2 file lists, the WAL) fills the NVMe: no claims below PROBE_MIN_FREE_GB. */
|
||
private space = new DiskSpace('probing', '/', coordinatorConfig.probe.minFreeGb)
|
||
private share = new NntpShare('prober', coordinatorConfig.nntpShares.proberPriority, coordinatorConfig.probe.minConnections, coordinatorConfig.probe.connections)
|
||
|
||
/** Per minute: what was read, and where the lanes' time went (ms summed over lanes; flushes ms on their own). */
|
||
private static freshStats() {
|
||
return { articles: 0, bytes: 0, done: 0, missing: 0, error: 0, cached: 0, waitMs: 0, idleMs: 0, fetchMs: 0, inspectMs: 0, flushMs: 0, claimMs: 0, since: Date.now() }
|
||
}
|
||
|
||
async run() {
|
||
// A previous run's claims: one prober at a time.
|
||
const { rowCount } = await db.rawQuery(`UPDATE probe_candidates SET status = 'waiting' WHERE status = 'taken'`)
|
||
if (rowCount) logger.info(`Probing: ${rowCount} candidates from a previous run back in the queue.`)
|
||
logger.info(`Probing with ${this.share.min}–${this.share.max} connections (its share of the account's).`)
|
||
const flusher = (async () => {
|
||
while (!this.stopped || this.outcomes.length > 0) {
|
||
await sleep(FLUSH_MS)
|
||
const flushed = Date.now()
|
||
await this.flush().catch((error) => logger.error(`Probing: storing results failed: ${error.message}`))
|
||
this.stats.flushMs += Date.now() - flushed
|
||
this.report()
|
||
}
|
||
})()
|
||
// Lanes as the share allows: the coordinator's history comes first (app/services/nntp_shares.ts).
|
||
while (!this.stopped) {
|
||
try {
|
||
const was = this.share.granted
|
||
const granted = await this.share.sync(this.dry ? 0 : this.share.max, this.running)
|
||
if (granted !== was) logger.info(`Probing: ${granted} connections (was ${was}).`)
|
||
while (this.running < granted) {
|
||
this.running++
|
||
const lane: Promise<void> = this.lane()
|
||
.catch((error) => logger.error(`Probing: a lane failed: ${error.message}`))
|
||
.finally(() => this.lanes.delete(lane))
|
||
this.lanes.add(lane)
|
||
}
|
||
} catch (error: any) {
|
||
logger.error(`Probing: NNTP share failed: ${error.message}`)
|
||
}
|
||
for (let waited = 0; waited < coordinatorConfig.nntpShares.everyMs && !this.stopped; waited += 1000) await sleep(1000)
|
||
}
|
||
await Promise.all(this.lanes)
|
||
await this.share.leave().catch(() => {})
|
||
// What was claimed and not probed goes back.
|
||
for (const candidate of this.queue.splice(0)) this.outcomes.push({ candidate, status: 'waiting' })
|
||
this.stopped = true
|
||
await flusher
|
||
}
|
||
|
||
stop() {
|
||
this.stopped = true
|
||
}
|
||
|
||
/**
|
||
* One connection's loop: PROBE_PIPELINE articles at a time (pipelined), reconnecting after a failure; done when
|
||
* there are more lanes than the share (its connection closed, the coordinator takes it on its next sync).
|
||
*/
|
||
private async lane() {
|
||
let client: XoverClient | null = null
|
||
let usedAt = Date.now()
|
||
try {
|
||
while (!this.stopped) {
|
||
if (this.running > this.share.granted) break
|
||
const batch: Candidate[] = []
|
||
const waited = Date.now()
|
||
while (batch.length < coordinatorConfig.probe.pipeline) {
|
||
const candidate = await this.next()
|
||
if (!candidate) break
|
||
batch.push(candidate)
|
||
}
|
||
this.stats.waitMs += Date.now() - waited
|
||
this.dry = batch.length === 0
|
||
if (batch.length === 0) {
|
||
// Nothing to probe for a while: the connection closes (nntpIdleCloseMs), and opens again with work.
|
||
const idleMs = coordinatorConfig.nntpIdleCloseMs
|
||
if (client && idleMs > 0 && Date.now() - usedAt >= idleMs) {
|
||
client.close()
|
||
client = null
|
||
}
|
||
this.stats.idleMs += 30_000
|
||
await sleep(30_000)
|
||
continue
|
||
}
|
||
try {
|
||
client ??= await XoverClient.connect({
|
||
host: env.get('NNTP_HOST')!,
|
||
port: env.get('NNTP_PORT', 563),
|
||
user: env.get('NNTP_USER'),
|
||
password: env.get('NNTP_PASSWORD'),
|
||
secure: env.get('NNTP_SECURE', true),
|
||
compress: false,
|
||
timeoutMs: coordinatorConfig.nntpTimeoutMs,
|
||
})
|
||
const fetched = Date.now()
|
||
const bodies = await client.bodies(batch.map((c) => c.messageId))
|
||
this.stats.fetchMs += Date.now() - fetched
|
||
const inspected = Date.now()
|
||
for (const [i, candidate] of batch.entries()) {
|
||
const body = bodies[i]
|
||
this.stats.articles++
|
||
if (body === null) {
|
||
// Taken down or expired: final.
|
||
this.outcomes.push({ candidate, status: 'missing' })
|
||
continue
|
||
}
|
||
this.stats.bytes += body.length
|
||
try {
|
||
this.outcomes.push({ candidate, status: 'done', inspection: inspect(body) })
|
||
} catch (error: any) {
|
||
// Not yEnc, a broken one, or data a parser chokes on: a final answer, not a connection problem.
|
||
if (!(error instanceof YencError)) logger.warn(`Probing ${candidate.messageId}: ${error.message}`)
|
||
this.outcomes.push({ candidate, status: 'error' })
|
||
}
|
||
}
|
||
this.stats.inspectMs += Date.now() - inspected
|
||
usedAt = Date.now()
|
||
} catch (error: any) {
|
||
logger.warn(`Probing ${batch[0].messageId} and ${batch.length - 1} more: ${error.message}`)
|
||
for (const candidate of batch) this.outcomes.push({ candidate, status: 'waiting' })
|
||
client?.close()
|
||
client = null
|
||
await sleep(5000)
|
||
}
|
||
}
|
||
} finally {
|
||
client?.close()
|
||
this.running--
|
||
}
|
||
}
|
||
|
||
/** Candidates claimed at a time: a few rounds of every lane's pipeline (a claim takes seconds; 400 lasted ~2 s at 37 lanes). */
|
||
private get batch() {
|
||
return Math.max(coordinatorConfig.probe.batch, this.running * coordinatorConfig.probe.pipeline * 4)
|
||
}
|
||
|
||
private async next(): Promise<Candidate | undefined> {
|
||
if (this.queue.length < this.batch / 2) {
|
||
this.refilling ??= this.claim().finally(() => (this.refilling = null))
|
||
if (this.queue.length === 0) await this.refilling
|
||
}
|
||
return this.queue.shift()
|
||
}
|
||
|
||
/**
|
||
* Waiting candidates of settled releases, by kind, claimed so another round
|
||
* doesn't take them: SLICES slices from random points of the queue. In id
|
||
* order it went group by group, and some groups' articles take 0.6–3 s
|
||
* where most take 20–60 ms (a.b.zogg 2016: 7 articles/s on 6 connections).
|
||
*/
|
||
private async claim() {
|
||
const started = Date.now()
|
||
try {
|
||
await this.claimTiers()
|
||
} finally {
|
||
this.stats.claimMs += Date.now() - started
|
||
}
|
||
}
|
||
|
||
private async claimTiers() {
|
||
// The queue empties, the lanes go dry and the share shrinks to its minimum until there's room again.
|
||
if (await this.space.short()) return
|
||
for (const { kind, unreadable } of TIERS) {
|
||
const want = this.batch - this.queue.length
|
||
if (want <= 0) return
|
||
// The starting points chosen here: in the query Postgres drew random() once for all slices.
|
||
// Over the claimable ones of this tier only: the rest of the queue isn't settled yet.
|
||
// Per value of unreadable, each an end of the index: over `unreadable >= false` Postgres walked every waiting
|
||
// entry of the kind, 2.5 minutes a claim at 28M archive candidates (LOG.md 2026-10-18 (74)).
|
||
const values = unreadable ? [true] : [false, true]
|
||
const ends = values
|
||
.map(() => `SELECT min(id) AS lo, max(id) AS hi FROM probe_candidates WHERE status = 'waiting' AND ready AND kind = ? AND unreadable = ?`)
|
||
.join(' UNION ALL ')
|
||
const [{ lo, hi }] = (
|
||
await db.rawQuery(
|
||
`SELECT min(lo) AS lo, max(hi) AS hi FROM (${ends}) e`,
|
||
values.flatMap((v) => [kind, v])
|
||
)
|
||
).rows
|
||
if (lo === null) continue
|
||
const starts = Array.from({ length: SLICES }, () => Number(lo) + Math.floor(Math.random() * (Number(hi) - Number(lo) + 1)))
|
||
const { rows } = await db.rawQuery(
|
||
`UPDATE probe_candidates SET status = 'taken' WHERE id IN (
|
||
SELECT pc.id FROM unnest(?::bigint[]) AS o(start)
|
||
CROSS JOIN LATERAL (
|
||
SELECT pc.id FROM probe_candidates pc
|
||
WHERE pc.status = 'waiting' AND pc.ready AND pc.kind = ? AND pc.unreadable >= ? AND pc.id >= o.start
|
||
ORDER BY pc.kind, pc.unreadable, pc.id LIMIT ? FOR UPDATE SKIP LOCKED) pc)
|
||
RETURNING id, release_id, kind, message_id`,
|
||
[starts, kind, unreadable, Math.ceil(want / SLICES)]
|
||
)
|
||
const claimed = rows.map((r: any) => ({ id: Number(r.id), releaseId: Number(r.release_id), kind: r.kind, messageId: r.message_id }))
|
||
// Probed before (an earlier catalog's candidate of the same article): answered from probe_cache.
|
||
const { rows: cached } = await db.rawQuery(`SELECT message_id, status, result FROM probe_cache WHERE message_id = ANY(?) AND version = ?`, [
|
||
claimed.map((c: Candidate) => c.messageId),
|
||
PROBE_VERSION,
|
||
])
|
||
const known = new Map(cached.map((c: any) => [c.message_id, c]))
|
||
for (const candidate of claimed) {
|
||
const hit: any = known.get(candidate.messageId)
|
||
if (hit) {
|
||
this.stats.cached++
|
||
this.outcomes.push({ candidate, status: hit.status, inspection: hit.result ?? undefined, cached: true })
|
||
} else {
|
||
this.queue.push(candidate)
|
||
}
|
||
}
|
||
}
|
||
}
|
||
|
||
/** The outcomes so far: candidates' status, inspections, and the releases' passworded and inner name. */
|
||
private async flush() {
|
||
const outcomes = this.outcomes.splice(0)
|
||
if (outcomes.length === 0) return
|
||
for (const o of outcomes) if (o.status !== 'waiting') this.stats[o.status]++
|
||
await db.transaction(async (trx) => {
|
||
await trx.rawQuery(
|
||
`UPDATE probe_candidates pc SET status = v.status FROM jsonb_to_recordset(?::jsonb) AS v(id bigint, status text) WHERE pc.id = v.id`,
|
||
[jsonb(outcomes.map((o) => ({ id: o.candidate.id, status: o.status })))]
|
||
)
|
||
// Final answers read from the news server, kept by message-id for the next catalog (a reset).
|
||
// One per message-id: the same article is a candidate of more than one release, and ON CONFLICT can't take a
|
||
// row twice in one statement.
|
||
const fresh = [...new Map(outcomes.filter((o) => !o.cached && o.status !== 'waiting').map((o) => [o.candidate.messageId, o])).values()]
|
||
if (fresh.length > 0) {
|
||
await trx.rawQuery(
|
||
`INSERT INTO probe_cache (message_id, status, result, version)
|
||
SELECT v.message_id, v.status, v.result, ? FROM jsonb_to_recordset(?::jsonb) AS v(message_id text, status text, result jsonb)
|
||
ON CONFLICT (message_id) DO UPDATE SET status = excluded.status, result = excluded.result, version = excluded.version, created_at = now()`,
|
||
[PROBE_VERSION, jsonb(fresh.map((o) => ({ message_id: o.candidate.messageId, status: o.status, result: o.inspection ?? null })))]
|
||
)
|
||
}
|
||
const done = outcomes.filter((o) => o.inspection)
|
||
if (done.length === 0) return
|
||
await trx.rawQuery(
|
||
`INSERT INTO release_inspections (release_id, candidate_id, kind, result)
|
||
SELECT v.release_id, v.candidate_id, v.kind, v.result FROM jsonb_to_recordset(?::jsonb) AS v(release_id bigint, candidate_id bigint, kind text, result jsonb)
|
||
ON CONFLICT (release_id, candidate_id) DO UPDATE SET result = excluded.result, created_at = now()`,
|
||
// A par2 set's file list goes to par2_files, not into every inspection.
|
||
[jsonb(done.map((o) => ({ release_id: o.candidate.releaseId, candidate_id: o.candidate.id, kind: o.candidate.kind, result: { ...o.inspection, par2: undefined } })))]
|
||
)
|
||
await this.storeForMerge(trx, done)
|
||
// Each touched release from all its inspections (earlier rounds may have stored others).
|
||
const ids = [...new Set(done.map((o) => o.candidate.releaseId))]
|
||
const { rows } = await trx.rawQuery(
|
||
`SELECT i.release_id, i.kind, i.result, r.name FROM release_inspections i JOIN releases r ON r.id = i.release_id WHERE i.release_id = ANY(?)`,
|
||
[ids]
|
||
)
|
||
const byRelease = new Map<number, { name: string; all: { kind: string; result: Inspection }[] }>()
|
||
for (const r of rows) {
|
||
const id = Number(r.release_id)
|
||
const entry = byRelease.get(id) ?? { name: r.name as string, all: [] as { kind: string; result: Inspection }[] }
|
||
entry.all.push({ kind: r.kind, result: r.result })
|
||
byRelease.set(id, entry)
|
||
}
|
||
const updates = [...byRelease].map(([id, { name, all }]) => ({
|
||
id,
|
||
passworded: all.some((i) => i.result.passworded === true),
|
||
inner_name: innerName(all, name),
|
||
}))
|
||
await trx.rawQuery(
|
||
`UPDATE releases r SET passworded = v.passworded, inner_name = v.inner_name, inspected_at = now()
|
||
FROM jsonb_to_recordset(?::jsonb) AS v(id bigint, passworded boolean, inner_name text) WHERE r.id = v.id`,
|
||
[jsonb(updates)]
|
||
)
|
||
})
|
||
}
|
||
|
||
/** What the par2 merge matches (app/services/par2_merger.ts): files' first-16-KiB hashes, par2 sets' file lists. */
|
||
private async storeForMerge(trx: any, done: Outcome[]) {
|
||
const hashed = done.filter((o) => o.inspection!.hash16k)
|
||
if (hashed.length > 0) {
|
||
await trx.rawQuery(
|
||
`INSERT INTO file_hashes (candidate_id, release_id, hash16k, size)
|
||
SELECT v.candidate_id, v.release_id, v.hash16k, v.size FROM jsonb_to_recordset(?::jsonb) AS v(candidate_id bigint, release_id bigint, hash16k text, size bigint)
|
||
ON CONFLICT (candidate_id) DO NOTHING`,
|
||
[jsonb(hashed.map((o) => ({ candidate_id: o.candidate.id, release_id: o.candidate.releaseId, hash16k: o.inspection!.hash16k, size: o.inspection!.size })))]
|
||
)
|
||
}
|
||
for (const o of done) {
|
||
const set = o.inspection!.par2
|
||
if (!set) continue
|
||
await trx.rawQuery(
|
||
`INSERT INTO par2_files (set_id, file_id, hash16k, length, name)
|
||
SELECT ?, v.file_id, v.hash16k, v.length, v.name FROM jsonb_to_recordset(?::jsonb) AS v(file_id text, hash16k text, length bigint, name text)
|
||
ON CONFLICT DO NOTHING`,
|
||
[set.setId, jsonb(set.files.map((f) => ({ file_id: f.fileId, hash16k: f.hash16k, length: f.length, name: f.name })))]
|
||
)
|
||
await trx.rawQuery(`INSERT INTO par2_sets (set_id, release_id, files) VALUES (?, ?, ?) ON CONFLICT DO NOTHING`, [set.setId, o.candidate.releaseId, set.files.length])
|
||
}
|
||
}
|
||
|
||
private report() {
|
||
const seconds = (Date.now() - this.stats.since) / 1000
|
||
if (seconds < REPORT_MS / 1000) return
|
||
const { articles, bytes, done, missing, error } = this.stats
|
||
logger.info(
|
||
`Probed ${articles} articles in ${seconds.toFixed(0)} s: ${(articles / seconds).toFixed(1)}/s, ${((bytes * 8) / seconds / 1e6).toFixed(0)} Mbit/s; ` +
|
||
`${done} inspected, ${missing} missing, ${error} not yEnc; ${this.stats.cached} from the probe cache. ` +
|
||
`Lanes: fetching ${(this.stats.fetchMs / 1000).toFixed(0)} s, waiting for candidates ${(this.stats.waitMs / 1000).toFixed(0)} s, ` +
|
||
`idle ${(this.stats.idleMs / 1000).toFixed(0)} s, inspecting ${(this.stats.inspectMs / 1000).toFixed(1)} s; ` +
|
||
`claims ${(this.stats.claimMs / 1000).toFixed(1)} s, flushes ${(this.stats.flushMs / 1000).toFixed(1)} s.`
|
||
)
|
||
this.stats = Prober.freshStats()
|
||
}
|
||
}
|