Rebuild spec: homelab measurements

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

REBUILD.md

@@ -29,6 +29,15 @@ search. Measured on the workstation (4 cores, 100 Mbit, Eweka):
   each; server side ~17% (coordinator) + ~8% (web) + ~3% (Postgres) of a
   core. (Before this round's fixes Postgres was at 32%: the merger's
   neighbour query read every range per file.)
+- On homelab (12 cores, 1 Gbit; 2026-10-09, separate containers and
+  database, `/mnt/bulk/usenet-rebuild`): 20 connections fetched ~630 k
+  articles/s (~31 k per connection, ~3 s per 100 k chunk, ~250 Mbit/s at
+  ~50 B per article on older boneless) and 8 worker processes indexed all of
+  it, waiting for the spool (~366 M headers in ~12 minutes). Server side at
+  that rate: coordinator 0.4–0.5 cores (1.3–1.7 before it stopped
+  inflating replies to check them), web 0.3–0.45, Postgres (on the HDD)
+  under 0.15. Worker per core there: ~195 k headers/s (boneless 2026),
+  ~275 k (ath 2013).
 - Storage: chunks 17.9–20.3 B per segment (format v2); a release ~1.1 KB.
 - For the whole target (528 B overview lines): ~36 TB of compressed XOVER
   to download. At 50 connections (~1–1.25 M lines/s, ~0.6–0.7 Gbit/s)

app/controllers/node_controller.ts

@@ -74,6 +74,12 @@ export default class NodeController {
     return this.lifecycle(ctx, (rangeId, nodeId) => Leases.finish(rangeId, nodeId))
   }
 
+  /** POST /ranges/:id/corrupt { from }: a chunk that doesn't inflate; the range goes back. */
+  async corrupt(ctx: HttpContext) {
+    const from = Number(ctx.request.input('from'))
+    return this.lifecycle(ctx, (rangeId, nodeId) => Leases.corrupt(rangeId, nodeId, from))
+  }
+
   async release(ctx: HttpContext) {
     return this.lifecycle(ctx, (rangeId, nodeId) => Leases.release(rangeId, nodeId))
   }

app/services/leases.ts

@@ -58,6 +58,18 @@ class Leases {
     await Spool.remove(rangeId)
   }
 
+  /**
+   * A worker got a chunk that doesn't inflate: the spooled chunk goes and
+   * the range back to the queue (the prefetcher fetches leased ranges'
+   * missing chunks again; the next worker resumes at the range's resume point).
+   */
+  async corrupt(rangeId: number, nodeId: number, from: number) {
+    const range = await this.leased(rangeId, nodeId)
+    const span = Spool.chunksOf({ start: Number(range.start_id), end: Number(range.end_id) }).find((s) => s.from === from)
+    if (span) await Spool.removeChunk(rangeId, span)
+    await db.from('ranges').where('id', rangeId).update({ status: 'planned', node_id: null })
+  }
+
   /** A stopping worker gives its range back; the next one resumes it. */
   async release(rangeId: number, nodeId: number) {
     await this.leased(rangeId, nodeId)

app/services/spool.ts

@@ -53,6 +53,11 @@ class Spool {
     return readFile(this.path(rangeId, span))
   }
 
+  /** One chunk, e.g. one that doesn't inflate (a false end of reply; see XoverClient.compressedRaw). */
+  async removeChunk(rangeId: number, span: ChunkSpan) {
+    await rm(this.path(rangeId, span), { force: true })
+  }
+
   async remove(rangeId: number) {
     await rm(join(this.dir, String(rangeId)), { recursive: true, force: true })
   }

commands/edge_node.ts

@@ -13,6 +13,8 @@ const gzipAsync = promisify(gzip)
 const RETRIES = 8
 
 class LeaseLost extends Error {}
+/** A spooled chunk that doesn't inflate (reported; the coordinator fetches it again). */
+class CorruptChunk extends Error {}
 
 /**
  * An edge worker (REBUILD.md, part 2): leases a range from the coordinator,
@@ -134,9 +136,17 @@ export default class EdgeNode extends BaseCommand {
       posting.catch((error) => (postError ??= error))
     }
     let postError: Error | null = null
-    const handle = async (chunk: { to: number; text: Promise<Buffer> }) => {
+    const handle = async (chunk: { from: number; to: number; text: Promise<Buffer> }) => {
       if (postError) throw postError
-      indexer.addText((await chunk.text).toString('utf8'), chunk.to + 1)
+      let text: Buffer
+      try {
+        text = await chunk.text
+      } catch (error: any) {
+        await posting.catch(() => {})
+        await api.corrupt(range.id, chunk.from)
+        throw new CorruptChunk(`chunk ${chunk.from}–${chunk.to} doesn't inflate (${error.message}); reported, the range goes back`)
+      }
+      indexer.addText(text.toString('utf8'), chunk.to + 1)
       reachedEnd = chunk.to >= range.end
       if (reachedEnd) indexer.finish()
       if (reachedEnd || chunk.to + 1 - batchFrom >= this.batchArticles) {
@@ -146,11 +156,11 @@ export default class EdgeNode extends BaseCommand {
     }
     try {
       // Inflating runs on a thread-pool thread: the next chunk inflates while this one is indexed.
-      let pending: { to: number; text: Promise<Buffer> } | null = null
+      let pending: { from: number; to: number; text: Promise<Buffer> } | null = null
       for await (const frame of decodeFrames(await api.stream(range.id))) {
         if (this.stopping) break
         if (isKeepalive(frame)) continue
-        const next = { to: frame.to, text: unzipAsync(frame.data) }
+        const next = { from: frame.from, to: frame.to, text: unzipAsync(frame.data) }
         next.text.catch(() => {})
         if (pending) await handle(pending)
         pending = next
@@ -172,7 +182,7 @@ export default class EdgeNode extends BaseCommand {
       )
     } catch (error: any) {
       await posting.catch(() => {})
-      if (error instanceof LeaseLost) {
+      if (error instanceof LeaseLost || error instanceof CorruptChunk) {
         this.logger.warning(`Range ${range.id}: ${error.message}`)
         return
       }
@@ -227,6 +237,11 @@ class Api {
     return { ...((await response.json()) as { releases: number; candidates: number }), bytes: body.length }
   }
 
+  async corrupt(rangeId: number, from: number) {
+    const response = await this.request(`ranges/${rangeId}/corrupt`, { method: 'POST', body: JSON.stringify({ from }) })
+    if (!response.ok) throw new Error(`corrupt: HTTP ${response.status}`)
+  }
+
   async call(rangeId: number, action: 'finish' | 'release' | 'heartbeat') {
     const response = await this.request(`ranges/${rangeId}/${action}`, { method: 'POST', body: '{}' })
     if (!response.ok) throw new Error(`${action}: HTTP ${response.status}`)

lib/nntp/xover_client.ts

@@ -233,20 +233,31 @@ export class XoverClient {
   }
 
   /**
-   * A zlib/gzip stream followed by a "." line: complete once it ends in
+   * A zlib/gzip stream followed by a "." line. The pieces are joined only
-   * ".\r\n" and inflates. The pieces are joined only then: joining them on
+   * once the data ends in ".\r\n": joining them on every arrival was
-   * every arrival was quadratic, ~15% of a prefetcher's time (LOG.md 2026-10-09).
+   * quadratic, ~15% of a prefetcher's time (LOG.md 2026-10-09).
+   *
+   * Raw (the coordinator's spool) it is complete then: inflating every reply
+   * to be sure cost the coordinator ~1.5 cores at ~630 k articles/s (mostly
+   * GC of the inflated text). ".\r\n" can also end a network read inside
+   * the compressed data (~1 in 16 M reads): that chunk doesn't inflate on
+   * the worker, which reports it and the chunk is fetched again (REBUILD.md,
+   * prefetch), and the rest of the reply breaks this connection's next
+   * command. With the text wanted, it must inflate.
    */
   private async compressedRaw(wantText: boolean): Promise<{ raw: Buffer; text: string }> {
     while (true) {
       if (this.length > DOT_LINE.length && this.endsWith(DOT_LINE)) {
         const all = this.buffer()
         const raw = all.subarray(0, all.length - DOT_LINE.length)
+        if (!wantText) {
+          this.take(all.length)
+          return { raw, text: '' }
+        }
         try {
-          // On a thread-pool thread: a chunk's ~25 MB of text took the coordinator's event loop ~60 ms.
           const inflated = await unzipAsync(raw)
           this.take(all.length)
-          return { raw: Buffer.from(raw), text: wantText ? inflated.toString('utf8') : '' }
+          return { raw, text: inflated.toString('utf8') }
         } catch {
           // ".\r\n" inside the compressed data: wait for more.
         }

start/routes.ts

@@ -51,6 +51,7 @@ router
         router.post('ranges/:id/heartbeat', [controllers.Node, 'heartbeat'])
         router.post('ranges/:id/finish', [controllers.Node, 'finish'])
         router.post('ranges/:id/release', [controllers.Node, 'release'])
+        router.post('ranges/:id/corrupt', [controllers.Node, 'corrupt'])
       })
       .prefix('node')
       .as('node')

tests/functional/node_api.spec.ts

@@ -84,4 +84,24 @@ test.group('Node API', (group) => {
     assert.equal(range.status, 'done')
     assert.isFalse(await Spool.has(rangeId, { from: 1, to: 100 }))
   })
+  test('a chunk that does not inflate: reported, removed from the spool, the range goes back', async ({ client, assert }) => {
+    const [g] = await db.table('groups').insert({ name: GROUP, priority: 1 }).returning('id')
+    const [r] = await db.table('ranges').insert({ group_id: g.id, start_id: 1, end_id: 300, next_id: 1, spooled_chunks: 3 }).returning('id')
+    const rangeId = Number(r.id)
+    for (const span of Spool.chunksOf({ start: 1, end: 300 })) await Spool.write(rangeId, span, chunk(span.from + 10))
+    const node = await Node.create({ name: 'n1' })
+    const token = (await Node.accessTokens.create(node)).value!.release()
+    ;(await client.post('/api/v1/node/lease').bearerToken(token)).assertStatus(200)
+
+    ;(await client.post(`/api/v1/node/ranges/${rangeId}/corrupt`).bearerToken(token).json({ from: 101 })).assertStatus(200)
+    assert.deepEqual(
+      await Promise.all(Spool.chunksOf({ start: 1, end: 300 }).map((span) => Spool.has(rangeId, span))),
+      [true, false, true]
+    )
+    const range = await db.from('ranges').where('id', rangeId).first()
+    assert.equal(range.status, 'planned')
+    assert.isNull(range.node_id)
+    // No longer this node's: a second report is refused.
+    ;(await client.post(`/api/v1/node/ranges/${rangeId}/corrupt`).bearerToken(token).json({ from: 101 })).assertStatus(409)
+  })
 })