Prober: claim from random points of the queue, not in id order

In id order the prober went group by group, and some groups' articles
take 0.6-3 s each (a.b.zogg 2016; repeated fetches the same, pipelining
4 deep no faster), where a sample across all years takes 20-60 ms for
most: 7 articles/s on 6 connections, against 67/s in its first minutes.
A claim now takes 20 slices from random points (353 ms per 200).
tools/body_timing.ts: BODY time for the oldest and newest candidates.

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

app/services/prober.ts

@@ -19,6 +19,8 @@ type Outcome = { candidate: Candidate; status: 'done' | 'missing' | 'error' | 'w
 /** 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
+/** Slices a claim takes from random points of the queue. */
+const SLICES = 20
 const REPORT_MS = 60_000
 
 /**
@@ -114,18 +116,27 @@ export class Prober {
     return this.queue.shift()
   }
 
-  /** Waiting candidates of settled releases, by kind, claimed so another round doesn't take them. */
+  /**
+   * 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() {
     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)
+           SELECT pc.id FROM (SELECT min(id) AS lo, max(id) AS hi FROM probe_candidates) b
+           CROSS JOIN LATERAL generate_series(1, ?) AS s
+           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 >= b.lo + floor(random() * (b.hi - b.lo + 1))::bigint + s * 0
+             ORDER BY pc.id LIMIT ? FOR UPDATE OF pc SKIP LOCKED) pc)
          RETURNING id, release_id, kind, message_id`,
-        [kind, want]
+        [SLICES, kind, 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 })
     }

tools/body_timing.ts

@@ -0,0 +1,37 @@
+/**
+ * BODY timing by article age: candidates' first articles, oldest and newest
+ * posted, fetched one at a time on one connection (status line, then the
+ * rest), with the release's posting date.
+ *
+ *   node --env-file=.env --import=@poppinss/ts-exec tools/body_timing.ts [n]
+ */
+// @ts-ignore pg has no types here
+import pg from 'pg'
+import { XoverClient } from '#lib/nntp/index'
+
+const n = Number(process.argv[2] ?? 8)
+const env = process.env
+const db = new pg.Client({ host: env.DB_HOST, port: Number(env.DB_PORT ?? 5432), user: env.DB_USER, password: env.DB_PASSWORD, database: env.DB_DATABASE })
+await db.connect()
+const pick = async (order: string) =>
+  (await db.query(`SELECT pc.message_id, r.posted_at FROM probe_candidates pc JOIN releases r ON r.id = pc.release_id WHERE pc.kind = 'lone' AND r.posted_at IS NOT NULL ORDER BY r.posted_at ${order} LIMIT $1`, [n])).rows
+const client = await XoverClient.connect({ host: env.NNTP_HOST!, port: Number(env.NNTP_PORT ?? 563), user: env.NNTP_USER, password: env.NNTP_PASSWORD, secure: (env.NNTP_SECURE ?? 'true') !== 'false', compress: false, timeoutMs: 120_000 })
+for (const [label, rows] of [['oldest', await pick('ASC')], ['newest', await pick('DESC')]] as const) {
+  let total = 0
+  for (const r of rows) {
+    const at = performance.now()
+    let size = 0
+    try {
+      size = (await client.body(r.message_id)).length
+    } catch (e: any) {
+      size = -1
+      console.log(`  ${e.message}`)
+    }
+    const ms = performance.now() - at
+    total += ms
+    console.log(`${label} ${new Date(r.posted_at).toISOString().slice(0, 10)} ${(ms / 1000).toFixed(2)} s ${(size / 1000).toFixed(0)} kB`)
+  }
+  console.log(`${label}: ${(total / rows.length / 1000).toFixed(2)} s an article`)
+}
+client.close()
+await db.end()