Rebuild: the coordinator spools replies without inflating them

On homelab (1 Gbit, 20 connections, ~630k articles/s) the coordinator
used 1.3-1.7 cores, mostly GC of the text it inflated only to know a
compressed reply was complete, plus a copy of every reply. A raw reply
is now complete when it ends in ".\r\n" (no copy): 0.4-0.5 cores at
the same rate.

".\r\n" can also end a network read inside the compressed data (~1 in
16M reads, a few dozen times over the whole target). Such a chunk
doesn't inflate on the worker: it reports it (POST ranges/:id/corrupt),
the spooled chunk is removed and the range goes back to the queue; the
prefetcher fetches it again. The rest of that reply breaks the
connection's next command, so the connection is replaced.

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

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