Daan MeijerandClaude Opus 5.5
d405b658c3
jsonb can't hold \u0000, and ON CONFLICT can't take a row twice in one statement: either failed the whole flush, its outcomes lost and its candidates taken until a restart. NULs are dropped from strings; the probe_cache insert takes each message-id once. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
362 lines
17 KiB
TypeScript
362 lines
17 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
|
||
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) {
|
||
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
|
||
} 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()
|
||
}
|
||
}
|