Rebuild: end-to-end fixes — spool accounting, keep-alive frames, resume point, overlapped worker

End to end (workstation, recent a.b.boneless ranges from the spool): a 2M
range in 10–12 s, 166k–202k headers/s per worker process (edge:bench on
one core: 215k).

- Prefetcher: spooled_at was set against a total counted from the range's
  resume point, which shrinks as the worker advances: the last chunks were
  never fetched and the worker waited. Now counted against the whole
  range, only new chunks counted, and leased ranges are always checked
  against the files. Planned ranges ahead are bounded (it had spooled ever
  more); the spool size is measured every 30 s (a stat per file).
- The stream sends an empty frame every 20 s while waiting for a chunk
  (Node's fetch gave up on a quiet body after 5 minutes).
- A range taken over midway: the first batch starts at the resume point
  (it started at the range's start and was refused).
- edge:node inflates the next chunk on the thread pool while indexing the
  current one, and doesn't wait for a post before indexing on (posts stay
  in order; a failed post stops the range).

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

app/controllers/node_controller.ts

@@ -4,10 +4,12 @@ import { setTimeout as sleep } from 'node:timers/promises'
 import Leases, { LeaseLost } from '#services/leases'
 import BatchStore, { StaleBatch } from '#services/batch_store'
 import Spool from '#services/spool'
-import { encodeFrame, fromWire, type WireBatch } from '#lib/edge/index'
+import { KEEPALIVE, encodeFrame, fromWire, type WireBatch } from '#lib/edge/index'
 
 /** How long the stream waits for a chunk the prefetcher hasn't spooled yet. */
 const CHUNK_WAIT_MS = 15 * 60 * 1000
+/** While waiting, an empty frame this often: an HTTP client gives up on a quiet body (Node's after 5 minutes). */
+const KEEPALIVE_MS = 20_000
 
 /**
  * The edge workers' API (REBUILD.md, parts 1–2): a lease, the range's
@@ -34,8 +36,14 @@ export default class NodeController {
     async function* frames() {
       for (const span of spans) {
         const waitUntil = Date.now() + CHUNK_WAIT_MS
+        let quietSince = Date.now()
         while (!(await Spool.has(rangeId, span))) {
           if (Date.now() > waitUntil) throw new Error(`Chunk ${span.from}–${span.to} of range ${rangeId} wasn't spooled in time`)
+          if (Date.now() - quietSince >= KEEPALIVE_MS) {
+            quietSince = Date.now()
+            yield encodeFrame(KEEPALIVE)
+            await Leases.heartbeat(rangeId, nodeId)
+          }
           await sleep(500)
         }
         yield encodeFrame({ ...span, data: await Spool.read(rangeId, span) })

app/services/prefetcher.ts

@@ -9,12 +9,14 @@ interface Target {
   group: string
   /** Chunks not spooled yet, in order. */
   missing: ChunkSpan[]
-  /** All of the range's chunks from its resume point. */
+  /** All of the range's chunks, from its start. */
   total: number
 }
 
 /** How often the targets are read again from the database. */
 const REFRESH_MS = 2000
+/** How often the spool's size is measured (a stat per file). */
+const SIZE_MS = 30_000
 
 /**
  * Keeps the news server busy ahead of the workers (REBUILD.md, prefetch).
@@ -31,6 +33,8 @@ export class Prefetcher {
   private running = 0
   private refreshedAt = 0
   private stopped = false
+  private spoolBytes = 0
+  private sizedAt = 0
 
   /** One round: refreshes the targets now and then, and puts idle connections to work. */
   async tick() {
@@ -65,7 +69,10 @@ export class Prefetcher {
           await client.group(target.group)
           return client.xoverRaw(span.from, span.to)
         })
+        const existed = await Spool.has(target.id, span)
         await Spool.write(target.id, span, data)
+        if (existed) continue
+        this.spoolBytes += data.length
         const { rows } = await db.rawQuery(
           `UPDATE ranges SET spooled_chunks = spooled_chunks + 1 WHERE id = ? RETURNING spooled_chunks`,
           [target.id]
@@ -94,20 +101,31 @@ export class Prefetcher {
   }
 
   private async readTargets(): Promise<Target[]> {
-    const rows = (q: any) =>
-      q.from('ranges as r').join('groups as g', 'g.id', 'r.group_id').whereNull('r.spooled_at').select('r.id', 'g.name as group', 'r.start_id', 'r.end_id', 'r.next_id')
-    // Leased ranges whose chunks aren't all there: a worker is waiting for them.
+    if (Date.now() - this.sizedAt >= SIZE_MS) {
+      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')
+    // 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')
-    const [{ leases }] = await db.from('ranges').where('status', 'leased').count('* as leases')
-    const ahead = Math.max(coordinatorConfig.spool.minAhead, coordinatorConfig.spool.aheadPerLease * Number(leases))
-    const room = (await Spool.size()) < coordinatorConfig.spool.maxBytes
-    const planned = room ? await rows(db.query()).where('r.status', 'planned').orderBy([{ column: 'r.priority' }, { column: 'r.id' }]).limit(ahead) : []
+    // 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 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).orderBy([{ column: 'r.priority' }, { column: 'r.id' }]).limit(ahead - Number(ready))
+      : []
     const targets: Target[] = []
-    for (const row of [...leased, ...planned]) {
-      const spans = Spool.chunksOf({ start: Number(row.start_id), end: Number(row.end_id) }, Number(row.next_id))
+    for (const row of [...leased, ...(room ? started : []), ...fresh]) {
+      const start = Number(row.start_id)
+      const end = Number(row.end_id)
+      const total = Spool.chunksOf({ start, end }).length
       const missing: ChunkSpan[] = []
-      for (const span of spans) if (!this.claimed.has(`${row.id}:${span.from}`) && !(await Spool.has(Number(row.id), span))) missing.push(span)
-      targets.push({ id: Number(row.id), group: row.group, missing, total: spans.length })
+      for (const span of Spool.chunksOf({ start, end }, Number(row.next_id))) {
+        if (!this.claimed.has(`${row.id}:${span.from}`) && !(await Spool.has(Number(row.id), span))) missing.push(span)
+      }
+      if (missing.length > 0) targets.push({ id: Number(row.id), group: row.group, missing, total })
     }
     return targets
   }

commands/edge_node.ts

@@ -1,10 +1,12 @@
 import { BaseCommand, flags } from '@adonisjs/core/ace'
 import type { CommandOptions } from '@adonisjs/core/types/ace'
 import { setTimeout as sleep } from 'node:timers/promises'
-import { gzipSync, unzipSync } from 'node:zlib'
-import { RangeIndexer, decodeFrames, toWire, type EdgeBatch, type Lease } from '#lib/edge/index'
+import { promisify } from 'node:util'
+import { gzipSync, unzip } from 'node:zlib'
+import { RangeIndexer, decodeFrames, isKeepalive, toWire, type EdgeBatch, type Lease } from '#lib/edge/index'
 
 const IDLE_MS = 30_000
+const unzipAsync = promisify(unzip)
 const RETRIES = 8
 
 class LeaseLost extends Error {}
@@ -66,31 +68,54 @@ export default class EdgeNode extends BaseCommand {
   private async runRange(api: Api, lease: Lease) {
     const { range, rules } = lease
     const started = Date.now()
-    const indexer = new RangeIndexer({ group: range.group, start: range.start, end: range.end, rules })
+    const indexer = new RangeIndexer({ group: range.group, start: range.start, end: range.end, resumeAt: range.next, rules })
     const totals = { headers: 0, releases: 0, candidates: 0, bytes: 0, postMs: 0 }
     let chunks = 0
-    const post = async (batch: EdgeBatch) => {
-      const at = Date.now()
-      const stored = await api.batch(range.id, batch)
-      totals.postMs += Date.now() - at
-      totals.bytes += stored.bytes
-      totals.headers += batch.headers
-      totals.releases += stored.releases
-      totals.candidates += stored.candidates
+    let reachedEnd = false
+    // Posts go in order, but the worker indexes on meanwhile (a post took 1–16 s; LOG.md 2026-10-09).
+    let posting: Promise<void> = Promise.resolve()
+    const post = (batch: EdgeBatch) => {
+      const previous = posting
+      posting = (async () => {
+        await previous
+        const at = Date.now()
+        const stored = await api.batch(range.id, batch)
+        totals.postMs += Date.now() - at
+        totals.bytes += stored.bytes
+        totals.headers += batch.headers
+        totals.releases += stored.releases
+        totals.candidates += stored.candidates
+      })()
+      // A failed post stops the range at the next chunk (a lost lease: nothing more will be stored).
+      posting.catch((error) => (postError ??= error))
+    }
+    let postError: Error | null = null
+    const handle = async (chunk: { to: number; text: Promise<Buffer> }) => {
+      if (postError) throw postError
+      indexer.addText((await chunk.text).toString('utf8'), chunk.to + 1)
+      chunks++
+      reachedEnd = chunk.to >= range.end
+      if (reachedEnd) indexer.finish()
+      if (reachedEnd || chunks % this.batchChunks === 0) post(indexer.take())
     }
     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
       for await (const frame of decodeFrames(await api.stream(range.id))) {
-        indexer.addText(unzipSync(frame.data).toString('utf8'), frame.to + 1)
-        chunks++
-        const last = frame.to >= range.end
-        if (last) indexer.finish()
-        if (last || chunks % this.batchChunks === 0) await post(indexer.take())
-        if (this.stopping && !last) {
-          // Back to the queue; the next worker resumes at the last batch.
-          if (chunks % this.batchChunks !== 0) this.logger.info(`Range ${range.id}: giving it back (${chunks % this.batchChunks} chunks since the last batch are read again).`)
-          await api.call(range.id, 'release')
-          return
-        }
+        if (isKeepalive(frame)) continue
+        const next = { to: frame.to, text: unzipAsync(frame.data) }
+        next.text.catch(() => {})
+        if (pending) await handle(pending)
+        pending = next
+        if (this.stopping) break
+      }
+      if (pending && !this.stopping) await handle(pending)
+      await posting
+      if (!reachedEnd) {
+        // Stopping: back to the queue; the next worker resumes at the last stored batch.
+        this.logger.info(`Range ${range.id}: giving it back.`)
+        await api.call(range.id, 'release')
+        return
       }
       await api.call(range.id, 'finish')
       const seconds = (Date.now() - started) / 1000
@@ -100,6 +125,7 @@ export default class EdgeNode extends BaseCommand {
           `posted ${(totals.bytes / 1e6).toFixed(1)} MB in ${(totals.postMs / 1000).toFixed(1)} s.`
       )
     } catch (error: any) {
+      await posting.catch(() => {})
       if (error instanceof LeaseLost) {
         this.logger.warning(`Range ${range.id}: ${error.message}`)
         return

lib/edge/index.ts

@@ -1,4 +1,4 @@
 export { RangeIndexer, type EdgeBatch, type ReleaseChunk, type DroppedChunk, type ProbeCandidate, type CandidateKind, type RuleCount, type Fragment, type RangeIndexerOptions } from './range_indexer.js'
-export { encodeFrame, decodeFrames, type Frame } from './stream.js'
+export { encodeFrame, decodeFrames, isKeepalive, KEEPALIVE, type Frame } from './stream.js'
 export { toWire, fromWire, type WireBatch } from './wire.js'
 export type { Lease } from './lease.js'

lib/edge/range_indexer.ts

@@ -83,6 +83,8 @@ export interface RangeIndexerOptions {
   /** Inclusive article numbers. */
   start: number
   end: number
+  /** Where this worker starts: the range's resume point (a range taken over midway); its first batch starts here. */
+  resumeAt?: number
   rules: Rules
   /** Chunks without a new part before an open file is given up (as a fragment near an edge). */
   idleChunks?: number
@@ -112,7 +114,7 @@ export class RangeIndexer {
     this.rules = new RuleSet(options.rules)
     this.idle = options.idleChunks ?? 5
     this.edge = options.edgeArticles ?? 50_000
-    this.batch = this.empty(options.start)
+    this.batch = this.empty(options.resumeAt ?? options.start)
   }
 
   /** One chunk of overview lines (parsed); `next` is the first article after it. */

lib/edge/stream.ts

@@ -11,6 +11,13 @@ export interface Frame {
 
 const HEADER = 20
 
+/** An empty frame, sent while the next chunk is still being fetched, so the response doesn't go quiet. */
+export const KEEPALIVE: Frame = { from: 0, to: 0, data: Buffer.alloc(0) }
+
+export function isKeepalive(frame: Frame) {
+  return frame.data.length === 0 && frame.from === 0
+}
+
 export function encodeFrame(frame: Frame): Buffer {
   const head = Buffer.alloc(HEADER)
   head.writeBigUInt64BE(BigInt(frame.from), 0)

tests/unit/edge/range_indexer.spec.ts

@@ -65,3 +65,11 @@ test.group('edge / RangeIndexer', () => {
     assert.lengthOf(batch.releases, 0)
   })
 })
+
+test.group('edge / taking over a range', () => {
+  test("a worker that resumes a range midway starts its first batch at the range's resume point", ({ assert }) => {
+    const indexer = new RangeIndexer({ group: GROUP, start: 1, end: 1_000_000, resumeAt: 500_001, rules })
+    indexer.add(post(500_001, 'A.Release.2020.mkv', 1), 600_000)
+    assert.deepEqual([indexer.take().from, indexer.take().from], [500_001, 600_000])
+  })
+})

tests/unit/edge/stream.spec.ts

@@ -0,0 +1,28 @@
+import { test } from '@japa/runner'
+import { KEEPALIVE, decodeFrames, encodeFrame, isKeepalive } from '#lib/edge/index'
+
+test.group('edge / stream frames', () => {
+  test('frames survive being split anywhere; keep-alive frames are recognised', async ({ assert }) => {
+    const whole = Buffer.concat([
+      encodeFrame({ from: 160_000_000_001, to: 160_000_050_000, data: Buffer.from('first') }),
+      encodeFrame(KEEPALIVE),
+      encodeFrame({ from: 160_000_050_001, to: 160_000_100_000, data: Buffer.from('second chunk') }),
+    ])
+    const pieces = [whole.subarray(0, 7), whole.subarray(7, 30), whole.subarray(30)]
+    const frames = []
+    for await (const frame of decodeFrames(pieces)) frames.push(frame)
+    assert.deepEqual(frames.map((f) => [f.from, f.to, f.data.toString(), isKeepalive(f)]), [
+      [160_000_000_001, 160_000_050_000, 'first', false],
+      [0, 0, '', true],
+      [160_000_050_001, 160_000_100_000, 'second chunk', false],
+    ])
+  })
+
+  test('a stream that ends inside a frame is an error', async ({ assert }) => {
+    const cut = encodeFrame({ from: 1, to: 2, data: Buffer.from('abc') }).subarray(0, 21)
+    await assert.rejects(async () => {
+      for await (const _ of decodeFrames([cut])) {
+      }
+    }, /inside a frame/)
+  })
+})