Prober: pipelined bodies, merge-first order, a probe cache across resets

- XoverClient.bodies: PROBE_PIPELINE (8) BODY commands per connection at
  once, replies read in order (one at a time it ran 3-10 articles/s on 6
  connections against a queue of 1M).
- Claims in tiers: lone files first, then par2 sets and archives of
  releases without a readable name (what the par2 merge needs), then those
  of readable releases.
- probe_cache: final answers by message-id (status, inspection with the
  16 KiB hash and par2 file list), PROBE_VERSION 2. A reset empties the
  probe queue, not the cache; recreated candidates of probed articles are
  answered from it without the news server.

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

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<void> | 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() }
   }
 }

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: {

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`)
+  }
+}

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

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<Buffer> {
     let searched = 0

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'

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

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(),

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', '<gone@test>', '<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('<one@test>')).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)