Rebuild: phase 2's prober reads the probe queue

probe:run (compose service prober, PROBE_CONNECTIONS, default 6) claims
waiting candidates of settled releases (lone, then par2, then archive),
fetches each article with BODY over connections of its own (no
compression) and inspects it: the yEnc name and size, the content type
from the first bytes, a par2 set's files, a RAR volume's entries and
whether they're encrypted. Results per release and kind in
release_inspections (CHUNK_TABLESPACE, the HDD); on the release
passworded, inner_name (the largest content name) and inspected_at.
430 = missing, not yEnc = error, both final; connection trouble puts
the candidate back.

Fetch and parse in one process: an article takes ~0.1 s per connection,
parsing a few ms (no queue in between). Parsers from voyager 1 (lib/yenc,
lib/rar, lib/par2, lib/content sniff) with their tests; XoverClient.body;
yencode.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

app/services/prober.ts

@@ -0,0 +1,183 @@
+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 { MissingArticle, XoverClient } from '#lib/nntp/index'
+import { YencError } from '#lib/yenc/index'
+import { inspect, innerName, type Inspection } from '#lib/probe/index'
+
+interface Candidate {
+  id: number
+  releaseId: number
+  kind: string
+  messageId: string
+}
+
+type Outcome = { candidate: Candidate; status: 'done' | 'missing' | 'error' | 'waiting'; inspection?: Inspection }
+
+/** Kinds in the order they're probed: a random name's real name first, then par2 sets, then archives. */
+const KINDS = ['lone', 'par2', 'archive']
+const FLUSH_MS = 2000
+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 = { articles: 0, bytes: 0, done: 0, missing: 0, error: 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.`)
+    const connections = coordinatorConfig.probe.connections
+    logger.info(`Probing with ${connections} connections.`)
+    const flusher = (async () => {
+      while (!this.stopped || this.outcomes.length > 0) {
+        await sleep(FLUSH_MS)
+        await this.flush().catch((error) => logger.error(`Probing: storing results failed: ${error.message}`))
+        this.report()
+      }
+    })()
+    await Promise.all(Array.from({ length: connections }, () => this.lane()))
+    // 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: an article at a time, reconnecting after a failure. */
+  private async lane() {
+    let client: XoverClient | null = null
+    while (!this.stopped) {
+      const candidate = await this.next()
+      if (!candidate) {
+        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 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' })
+        }
+      } 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' })
+        client?.close()
+        client = null
+        await sleep(5000)
+      }
+    }
+    client?.close()
+  }
+
+  private async next(): Promise<Candidate | undefined> {
+    if (this.queue.length < coordinatorConfig.probe.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. */
+  private async claim() {
+    for (const kind of KINDS) {
+      const want = coordinatorConfig.probe.batch - this.queue.length
+      if (want <= 0) return
+      const { rows } = await db.rawQuery(
+        `UPDATE probe_candidates SET status = 'taken' WHERE id IN (
+           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
+           ORDER BY pc.id LIMIT ? FOR UPDATE OF pc SKIP LOCKED)
+         RETURNING id, release_id, kind, message_id`,
+        [kind, want]
+      )
+      for (const r of rows) this.queue.push({ id: Number(r.id), releaseId: Number(r.release_id), kind: r.kind, messageId: r.message_id })
+    }
+  }
+
+  /** 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`,
+        [JSON.stringify(outcomes.map((o) => ({ id: o.candidate.id, status: o.status })))]
+      )
+      const done = outcomes.filter((o) => o.inspection)
+      if (done.length === 0) return
+      await trx.rawQuery(
+        `INSERT INTO release_inspections (release_id, kind, result)
+         SELECT v.release_id, v.kind, v.result FROM jsonb_to_recordset(?::jsonb) AS v(release_id bigint, kind text, result jsonb)
+         ON CONFLICT (release_id, kind) DO UPDATE SET result = excluded.result, created_at = now()`,
+        [JSON.stringify(done.map((o) => ({ release_id: o.candidate.releaseId, kind: o.candidate.kind, result: o.inspection })))]
+      )
+      // Each touched release from all its inspections (earlier rounds may have stored its other kinds).
+      const ids = [...new Set(done.map((o) => o.candidate.releaseId))]
+      const { rows } = await trx.rawQuery(`SELECT release_id, kind, result FROM release_inspections WHERE release_id = ANY(?)`, [ids])
+      const byRelease = new Map<number, Record<string, Inspection>>()
+      for (const r of rows) {
+        const id = Number(r.release_id)
+        byRelease.set(id, { ...byRelease.get(id), [r.kind]: r.result })
+      }
+      const updates = [...byRelease].map(([id, all]) => ({
+        id,
+        passworded: Object.values(all).some((i) => i.passworded === true),
+        inner_name: innerName(all),
+      }))
+      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`,
+        [JSON.stringify(updates)]
+      )
+    })
+  }
+
+  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 = { articles: 0, bytes: 0, done: 0, missing: 0, error: 0, since: Date.now() }
+  }
+}