diff --git a/app/services/prober.ts b/app/services/prober.ts index e073f56..01328e3 100644 --- a/app/services/prober.ts +++ b/app/services/prober.ts @@ -3,9 +3,9 @@ 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 { MissingArticle, XoverClient } from '#lib/nntp/index' +import { XoverClient } from '#lib/nntp/index' import { YencError } from '#lib/yenc/index' -import { inspect, innerName, type Inspection } from '#lib/probe/index' +import { inspect, innerName, PROBE_VERSION, type Inspection } from '#lib/probe/index' interface Candidate { id: number @@ -14,10 +14,19 @@ interface Candidate { messageId: string } -type Outcome = { candidate: Candidate; status: 'done' | 'missing' | 'error' | 'waiting'; inspection?: Inspection } +type Outcome = { candidate: Candidate; status: 'done' | 'missing' | 'error' | 'waiting'; inspection?: Inspection; cached?: boolean } -/** Kinds in the order they're probed: a random name's real name first, then par2 sets, then archives. */ -const KINDS = ['lone', 'par2', 'archive'] +/** + * 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 /** Slices a claim takes from random points of the queue. */ const SLICES = 20 @@ -36,7 +45,7 @@ export class Prober { private outcomes: Outcome[] = [] private refilling: Promise | null = null private stopped = false - private stats = { articles: 0, bytes: 0, done: 0, missing: 0, error: 0, since: Date.now() } + private stats = { articles: 0, bytes: 0, done: 0, missing: 0, error: 0, cached: 0, since: Date.now() } async run() { // A previous run's claims: one prober at a time. @@ -62,12 +71,17 @@ export class Prober { this.stopped = true } - /** One connection's loop: an article at a time, reconnecting after a failure. */ + /** One connection's loop: PROBE_PIPELINE articles at a time (pipelined), reconnecting after a failure. */ private async lane() { let client: XoverClient | null = null while (!this.stopped) { - const candidate = await this.next() - if (!candidate) { + const batch: Candidate[] = [] + while (batch.length < coordinatorConfig.probe.pipeline) { + const candidate = await this.next() + if (!candidate) break + batch.push(candidate) + } + if (batch.length === 0) { await sleep(30_000) continue } @@ -81,25 +95,27 @@ export class Prober { compress: false, timeoutMs: coordinatorConfig.nntpTimeoutMs, }) - const body = await client.body(candidate.messageId) - this.stats.articles++ - 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' }) + const bodies = await client.bodies(batch.map((c) => c.messageId)) + 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' }) + } } } catch (error: any) { - if (error instanceof MissingArticle) { - // Taken down or expired: final (Eweka takes ~450 ms per missing article). - this.stats.articles++ - this.outcomes.push({ candidate, status: 'missing' }) - continue - } - logger.warn(`Probing ${candidate.messageId}: ${error.message}`) - this.outcomes.push({ candidate, status: 'waiting' }) + 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) @@ -123,7 +139,7 @@ export class Prober { * where most take 20–60 ms (a.b.zogg 2016: 7 articles/s on 6 connections). */ private async claim() { - for (const kind of KINDS) { + for (const { kind, unreadable } of TIERS) { const want = coordinatorConfig.probe.batch - this.queue.length if (want <= 0) return // The starting points chosen here: in the query Postgres drew random() once for all slices. @@ -134,12 +150,27 @@ export class Prober { SELECT pc.id FROM unnest(?::bigint[]) AS o(start) CROSS JOIN LATERAL ( SELECT pc.id FROM probe_candidates pc JOIN releases r ON r.id = pc.release_id - WHERE pc.status = 'waiting' AND pc.kind = ? AND r.settled_at IS NOT NULL AND pc.id >= o.start + WHERE pc.status = 'waiting' AND pc.kind = ? AND r.settled_at IS NOT NULL AND (NOT ? OR NOT r.readable) AND pc.id >= o.start ORDER BY pc.id LIMIT ? FOR UPDATE OF pc SKIP LOCKED) pc) RETURNING id, release_id, kind, message_id`, - [starts, kind, Math.ceil(want / SLICES)] + [starts, kind, unreadable, Math.ceil(want / SLICES)] ) - for (const r of rows) this.queue.push({ id: Number(r.id), releaseId: Number(r.release_id), kind: r.kind, messageId: r.message_id }) + 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) + } + } } } @@ -153,6 +184,16 @@ export class Prober { `UPDATE probe_candidates pc SET status = v.status FROM jsonb_to_recordset(?::jsonb) AS v(id bigint, status text) WHERE pc.id = v.id`, [JSON.stringify(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). + const fresh = outcomes.filter((o) => !o.cached && o.status !== 'waiting') + 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, JSON.stringify(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( @@ -219,8 +260,8 @@ export class Prober { 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.` + `${done} inspected, ${missing} missing, ${error} not yEnc; ${this.stats.cached} from the probe cache.` ) - this.stats = { articles: 0, bytes: 0, done: 0, missing: 0, error: 0, since: Date.now() } + this.stats = { articles: 0, bytes: 0, done: 0, missing: 0, error: 0, cached: 0, since: Date.now() } } } diff --git a/config/coordinator.ts b/config/coordinator.ts index dc27def..c3528ab 100644 --- a/config/coordinator.ts +++ b/config/coordinator.ts @@ -73,7 +73,9 @@ export default { */ connections: env.get('PROBE_CONNECTIONS', 6), /** Candidates claimed at a time. */ - batch: 200, + batch: 400, + /** BODY commands in flight per connection (pipelined; XoverClient.bodies). */ + pipeline: env.get('PROBE_PIPELINE', 8), }, /** Stored XOVER chunks of ranges marked `capture`, for the grouping sweeps (app/services/corpus.ts). Off when unset. */ corpus: { diff --git a/database/migrations/1770000002300_probe_cache.ts b/database/migrations/1770000002300_probe_cache.ts new file mode 100644 index 0000000..c654505 --- /dev/null +++ b/database/migrations/1770000002300_probe_cache.ts @@ -0,0 +1,25 @@ +import { BaseSchema } from '@adonisjs/lucid/schema' + +/** + * What a probed article showed, by message-id (app/services/prober.ts): an + * article's inspection doesn't depend on the grouping rules, so a catalog + * reset (which empties the probe queue) needn't read it from the news server + * again. `version` is PROBE_VERSION when it was inspected: a change to the + * inspection makes older rows stale. Not truncated by a reset. + */ +export default class extends BaseSchema { + async up() { + this.schema.raw(` + CREATE TABLE probe_cache ( + message_id text PRIMARY KEY, + status varchar(8) NOT NULL, + result jsonb, + version integer NOT NULL, + created_at timestamptz NOT NULL DEFAULT now() + )`) + } + + async down() { + this.schema.raw(`DROP TABLE IF EXISTS probe_cache`) + } +} diff --git a/database/schema.ts b/database/schema.ts index 4380e60..52d0fd3 100644 --- a/database/schema.ts +++ b/database/schema.ts @@ -153,6 +153,21 @@ export class Par2SetSchema extends BaseModel { declare setId: string } +export class ProbeCacheSchema extends BaseModel { + static $columns = ['createdAt', 'messageId', 'result', 'status', 'version'] as const + $columns = ProbeCacheSchema.$columns + @column.dateTime({ autoCreate: true }) + declare createdAt: DateTime + @column({ isPrimary: true }) + declare messageId: string + @column() + declare result: any | null + @column() + declare status: string + @column() + declare version: number +} + export class ProbeCandidateSchema extends BaseModel { static $columns = ['createdAt', 'file', 'id', 'kind', 'messageId', 'releaseId', 'status'] as const $columns = ProbeCandidateSchema.$columns diff --git a/lib/nntp/xover_client.ts b/lib/nntp/xover_client.ts index 7e2bbf1..9db1c71 100644 --- a/lib/nntp/xover_client.ts +++ b/lib/nntp/xover_client.ts @@ -246,6 +246,28 @@ export class XoverClient { }) } + /** + * Several bodies, pipelined: every BODY is sent at once and the replies read in order, so one round trip + * covers them all (one at a time, a connection waited ~0.1–2 s per article: 3–10 articles/s on 6; LOG.md + * 2026-10-12 (16)). A missing article (430, 423) is null. + */ + async bodies(messageIds: string[]): Promise<(Buffer | null)[]> { + return this.timed(async () => { + this.sock.write(messageIds.map((m) => `BODY ${m.startsWith('<') ? m : `<${m}>`}\r\n`).join('')) + const out: (Buffer | null)[] = [] + for (let i = 0; i < messageIds.length; i++) { + const reply = await this.line() + if (reply.startsWith('430') || reply.startsWith('423')) { + out.push(null) + continue + } + await this.expect(reply, ['222']) + out.push(await this.rawBody()) + } + return out + }) + } + /** A multi-line body up to "\r\n.\r\n", as bytes, the terminator cut off. */ private async rawBody(): Promise { let searched = 0 diff --git a/lib/probe/index.ts b/lib/probe/index.ts index 79351c2..51cc1f1 100644 --- a/lib/probe/index.ts +++ b/lib/probe/index.ts @@ -1 +1 @@ -export { inspect, innerName, MAX_NAMES, type Inspection } from './inspect.js' +export { inspect, innerName, MAX_NAMES, PROBE_VERSION, type Inspection } from './inspect.js' diff --git a/lib/probe/inspect.ts b/lib/probe/inspect.ts index 05a702a..4af1a97 100644 --- a/lib/probe/inspect.ts +++ b/lib/probe/inspect.ts @@ -28,6 +28,9 @@ export interface Inspection { const HASH_BYTES = 16 * 1024 +/** The inspection's version: raised when inspect() changes, so probe_cache rows of an older one are read again. */ +export const PROBE_VERSION = 2 + /** Names kept per inspection: the catalog wants the content's name, not every volume's. */ export const MAX_NAMES = 8 const MAX_NAME_LENGTH = 200 diff --git a/start/env.ts b/start/env.ts index 5bada5a..7667e86 100644 --- a/start/env.ts +++ b/start/env.ts @@ -61,6 +61,7 @@ export default await Env.create(new URL('../', import.meta.url), { LEASE_TTL_SECONDS: Env.schema.number.optional(), SPOOL_DIR: Env.schema.string.optional(), CORPUS_DIR: Env.schema.string.optional(), + PROBE_PIPELINE: Env.schema.number.optional(), SPOOL_AHEAD_PER_LEASE: Env.schema.number.optional(), SPOOL_MIN_AHEAD: Env.schema.number.optional(), SPOOL_MAX_GB: Env.schema.number.optional(), diff --git a/tests/unit/nntp/xover_client.spec.ts b/tests/unit/nntp/xover_client.spec.ts index a651329..0ab5b0e 100644 --- a/tests/unit/nntp/xover_client.spec.ts +++ b/tests/unit/nntp/xover_client.spec.ts @@ -79,6 +79,23 @@ test.group('nntp / XoverClient', () => { } }) + test('reads several bodies pipelined, in order, a missing one as null', async ({ assert }) => { + const { port, close } = await server() + const client = await XoverClient.connect({ host: '127.0.0.1', port, user: 'u', password: 'p', compress: false }) + try { + const bodies = await client.bodies(['one@test', '', '']) + assert.equal(bodies.length, 3) + assert.equal(bodies[0]!.subarray(6).toString('latin1'), '..stuffed\r\nlast\r\n') + assert.isNull(bodies[1]) + assert.equal(bodies[2]!.length, bodies[0]!.length) + // The connection is still in step afterwards. + assert.equal((await client.body('')).length, bodies[0]!.length) + } finally { + client.close() + close() + } + }) + test('parses overview lines into the fields nntp-js gives', ({ assert }) => { const [[number, fields]] = parseOverviews(LINES[0] + '\r\n') assert.equal(number, 101)