Rebuild: coordinator:run --shard=i/n splits prefetching over processes

The coordinator's event loop was 16.5% busy at ~10 MB/s of compressed
XOVER (socket reads, buffers, GC, spool writes; nothing stands out), so
one process tops out around 60 MB/s: less than 50 connections bring
(~1M articles/s, ~70-86 MB/s). A shard prefetches the ranges with
id % n = i - 1 with its own NNTP_CONNECTIONS; only shard 1 plans, merges
and settles. Two shards of 5 connections fetched what one of 10 did.

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

REBUILD.md

@@ -46,14 +46,14 @@ dropping garbage; 28 TB of compressed overview traffic. Throughput goal:
    account; the gzip stream is passed on as is).
 
    **Prefetch**: the coordinator fetches the next ranges ahead of the
-   workers, in chunks of 50 k lines and in order, into a spool on disk
+   workers, in chunks of 100 k lines and in order, into a spool on disk
    (~106 MB per 2 M range at ~53 B per line; on the HDD, written and read
    sequentially). It keeps about two ranges ahead per active lease, capped
    by size (default 10 GB), and fetches with all its header connections all
    the time, independent of how fast workers are. A lease goes to a range
-   that is already (partly) spooled; a worker reads its chunks from the
-   spool, waiting only for chunks still being fetched, and falls back to a
-   direct relay fetch for a range that isn't spooled. A worker gets its
+   that is already (partly) spooled, else to the next planned one (leased
+   ranges are fetched first); a worker reads its chunks from the spool,
+   waiting only for chunks still being fetched. A worker gets its
    whole range in one streaming request (`GET /ranges/:id/stream`): the
    spooled chunks in order, waiting for those still being fetched. It still
    parses chunk by chunk (2 M lines are ~560 MB of text) and posts results
@@ -64,6 +64,11 @@ dropping garbage; 28 TB of compressed overview traffic. Throughput goal:
    range is leased again later.
    Order: oldest era first (2008–2018 holds 63% of the readable files on
    17% of the lines), then forward.
+   One coordinator process handles ~60 MB/s of compressed XOVER (its event
+   loop was 16.5% busy at ~10 MB/s); beyond that, prefetching is split over
+   processes with `coordinator:run --shard=i/n` (ranges with id % n = i − 1,
+   each with its own NNTP_CONNECTIONS; only shard 1 plans, merges and
+   settles).
 
 2. **Edge worker** (node, TypeScript reusing `lib/`). Measured (`edge:bench`,
    2026-10-09): 73 k headers/s per core on an a.b.ath range from 2013,

app/services/prefetcher.ts

@@ -28,6 +28,13 @@ const SIZE_MS = 30_000
  * kept its worker waiting (LOG.md 2026-10-09).
  */
 export class Prefetcher {
+  /**
+   * This process's share of the ranges, `id % count = index` (`coordinator:run --shard`): one
+   * process's event loop handles ~60 MB/s of compressed XOVER (16.5% of a core at ~10 MB/s;
+   * LOG.md 2026-10-09), less than 50 connections bring.
+   */
+  constructor(private shard = { index: 0, count: 1 }) {}
+
   private targets: Target[] = []
   private claimed = new Set<string>()
   private running = 0
@@ -126,13 +133,14 @@ export class Prefetcher {
       this.sizedAt = Date.now()
       this.spoolBytes = await Spool.size()
     }
-    const rows = (q: any) => q.from('ranges as r').join('groups as g', 'g.id', 'r.group_id').select('r.id', 'g.name as group', 'r.start_id', 'r.end_id', 'r.next_id')
+    const mine = (q: any) => (this.shard.count > 1 ? q.whereRaw('r.id % ? = ?', [this.shard.count, this.shard.index]) : q)
+    const rows = (q: any) => mine(q.from('ranges as r').join('groups as g', 'g.id', 'r.group_id').select('r.id', 'g.name as group', 'r.start_id', 'r.end_id', 'r.next_id'))
     // Leased ranges, checked against the spool's files whatever spooled_at says: a worker waits for them.
     const leased = await rows(db.query()).where('r.status', 'leased').orderBy('r.leased_at')
     // Planned ranges ahead: those started already, then new ones up to SPOOL_AHEAD_PER_LEASE per lease.
     const ahead = Math.max(coordinatorConfig.spool.minAhead, coordinatorConfig.spool.aheadPerLease * leased.length)
     const started = await rows(db.query()).where('r.status', 'planned').where('r.spooled_chunks', '>', 0).whereNull('r.spooled_at').orderBy([{ column: 'r.priority' }, { column: 'r.id' }])
-    const [{ ready }] = await db.from('ranges').where('status', 'planned').where('spooled_chunks', '>', 0).count('* as ready')
+    const [{ ready }] = await mine(db.from('ranges as r').where('r.status', 'planned').where('r.spooled_chunks', '>', 0)).count('* as ready')
     const room = this.spoolBytes < coordinatorConfig.spool.maxBytes
     const fresh = room && Number(ready) < ahead
       ? await rows(db.query()).where('r.status', 'planned').where('r.spooled_chunks', 0).whereRaw('r.next_id <= r.end_id').orderBy([{ column: 'r.priority' }, { column: 'r.id' }]).limit(ahead - Number(ready))

commands/coordinator_run.ts

@@ -1,4 +1,4 @@
-import { BaseCommand } from '@adonisjs/core/ace'
+import { BaseCommand, flags } from '@adonisjs/core/ace'
 import type { CommandOptions } from '@adonisjs/core/types/ace'
 import { setTimeout as sleep } from 'node:timers/promises'
 import { Prefetcher } from '#services/prefetcher'
@@ -24,8 +24,14 @@ export default class CoordinatorRun extends BaseCommand {
   static description = 'Prefetches ranges into the spool and plans new ones'
   static options: CommandOptions = { startApp: true, staysAlive: true }
 
+  @flags.string({ description: 'This process\'s share of the prefetching, "i/n" (ranges with id % n = i - 1); only 1/n plans, merges and settles' })
+  declare shard?: string
+
   async run() {
-    const prefetcher = new Prefetcher()
+    const [index, count] = (this.shard ?? '1/1').split('/').map(Number)
+    if (!(count >= 1 && index >= 1 && index <= count)) throw new Error(`--shard: "i/n" with 1 <= i <= n, not "${this.shard}"`)
+    const prefetcher = new Prefetcher({ index: index - 1, count })
+    const main = index === 1
     let stopping = false
     let stopped: () => void
     const done = new Promise<void>((resolve) => (stopped = resolve))
@@ -36,7 +42,7 @@ export default class CoordinatorRun extends BaseCommand {
       prefetcher.stop()
       await done
     })
-    this.logger.info('Coordinator started.')
+    this.logger.info(count > 1 ? `Coordinator started (shard ${index}/${count}${main ? '' : ', prefetching only'}).` : 'Coordinator started.')
     // Two loops: the prefetcher's ticks don't wait for merging or planning (a merge round
     // of up to MERGE_BUDGET_MS left the connections idle once their targets were fetched).
     const prefetching = (async () => {
@@ -53,14 +59,14 @@ export default class CoordinatorRun extends BaseCommand {
     let mergedAt = 0
     while (!stopping) {
       try {
-        if (Date.now() - plannedAt >= PLAN_EVERY_MS) {
+        if (main && Date.now() - plannedAt >= PLAN_EVERY_MS) {
           plannedAt = Date.now()
           const { groups, ranges } = await Planner.planAll()
           if (ranges > 0) this.logger.info(`Planned ${ranges} ranges over ${groups} groups.`)
           const settled = await Settler.run()
           if (settled > 0) this.logger.info(`Settled ${settled} releases.`)
         }
-        if (Date.now() - mergedAt >= MERGE_EVERY_MS) {
+        if (main && Date.now() - mergedAt >= MERGE_EVERY_MS) {
           mergedAt = Date.now()
           // Rounds until a whole pass over the file keys decides nothing more, or MERGE_BUDGET_MS.
           // (Going on while only waiting files were left kept Postgres busy all the time.)