The range stream polls the corpus as well as the spool while it waits

The corpus fill claims chunks of a leased range ahead of the leased run (the
prefetcher's claim keys are per range and chunk), stores them in the corpus
only, and the run leaves them there. The stream looked at the corpus once per
chunk and then polled the spool only: when the store came seconds after that
look, the stream waited its whole hour and the worker lost the group and its
open batch (three groups on 2026-10-27, LOG.md (118)). Now the wait loop checks
both.

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>

REBUILD.md

@@ -551,6 +551,13 @@ releases, a third of them one poster's image dump (27.7 M rows, LOG.md
    when the window was marked, planned again by the reset): hence
    `LEASE_CAPTURE_ONLY` (the lease, next and the prefetcher's run-ahead
    take capture ranges only) and the one-off repair of the priorities.
+   The stream polls the corpus as well as the spool while it waits for a
+   chunk: the corpus fill claims chunks of the same range ahead of the
+   leased run (the prefetcher's claims are per range and chunk), stores
+   them in the corpus only, and the run leaves them there; a stream that
+   looked at the corpus once, seconds before that store, waited its whole
+   hour on the spool and the worker lost the group and its open batch
+   (three groups on the first morning).
 
 6. **The corpus stays as it is.** Measured on 300 unfiltered chunks
    (30 M lines, LOG.md (102)): 43% of lines lack a "(k/N)" counter, and of

app/controllers/node_controller.ts

@@ -5,7 +5,7 @@ import db from '@adonisjs/lucid/services/db'
 import Leases, { LeaseLost } from '#services/leases'
 import GroupLeases from '#services/group_leases'
 import BatchStore, { StaleBatch } from '#services/batch_store'
-import Spool from '#services/spool'
+import Spool, { type ChunkSpan } from '#services/spool'
 import Corpus from '#services/corpus'
 import { KEEPALIVE, encodeFrame, fromWire, type WireBatch } from '#lib/edge/index'
 
@@ -99,19 +99,22 @@ export default class NodeController {
     const spans = Spool.chunksOf({ start: Number(range.start_id), end: Number(range.end_id) }, Number(range.next_id))
     const group = range.capture ? await db.from('groups').where('id', range.group_id).select('name').first() : null
     const corpusRange = group ? { group: group.name as string, start: Number(range.start_id), end: Number(range.end_id) } : null
+    // The chunk from the corpus or the spool, whichever has it. Both are polled while waiting: the corpus fill
+    // claims a chunk ahead of the leased run (the prefetcher's claims are per range and chunk), stores it in the
+    // corpus only, and the run then leaves it there — a stream that looked at the corpus once, before that
+    // store, waited the whole hour on the spool (three groups on 2026-10-27, LOG.md (118)).
+    const stored = async (span: ChunkSpan) =>
+      corpusRange && (await Corpus.has(corpusRange, span)) ? Corpus.read(corpusRange, span) : null
     async function* frames() {
       for (const span of spans) {
-        if (corpusRange && (await Corpus.has(corpusRange, span))) {
-          const data = await Corpus.read(corpusRange, span)
-          if (data) {
-            yield encodeFrame({ ...span, data })
-            await Leases.heartbeat(rangeId, nodeId)
-            continue
-          }
-        }
         const waitUntil = Date.now() + CHUNK_WAIT_MS
         let quietSince = Date.now()
-        while (!(await Spool.has(rangeId, span))) {
+        let data = await stored(span)
+        while (!data) {
+          if (await Spool.has(rangeId, span)) {
+            data = await Spool.read(rangeId, span)
+            break
+          }
           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()
@@ -119,8 +122,9 @@ export default class NodeController {
             await Leases.heartbeat(rangeId, nodeId)
           }
           await sleep(500)
+          data = await stored(span)
         }
-        yield encodeFrame({ ...span, data: await Spool.read(rangeId, span) })
+        yield encodeFrame({ ...span, data })
         // Streaming keeps the lease alive.
         await Leases.heartbeat(rangeId, nodeId)
       }

tests/functional/node_api.spec.ts

@@ -153,6 +153,38 @@ test.group('Node API', (group) => {
     }
   })
 
+  test('a chunk the corpus fill stores while the stream waits for it comes from the corpus', async ({ client, assert }) => {
+    const corpus = await mkdtemp(join(tmpdir(), 'corpus-'))
+    const saved = coordinatorConfig.corpus.dir
+    coordinatorConfig.corpus.dir = corpus
+    try {
+      const [g] = await db.table('groups').insert({ name: GROUP, priority: 1 }).returning('id')
+      // A capture range not in the corpus: its first chunk spooled by the leased run, the second claimed by
+      // the corpus fill, which stores it in the corpus only, after the stream started.
+      const [r] = await db.table('ranges').insert({ group_id: g.id, start_id: 1, end_id: 200, next_id: 1, capture: true }).returning('id')
+      const rangeId = Number(r.id)
+      await Spool.write(rangeId, { from: 1, to: 100 }, chunk(11))
+      const node = await Node.create({ name: 'n1' })
+      const token = (await Node.accessTokens.create(node)).value!.release()
+      await db.from('ranges').where('id', rangeId).update({ status: 'leased', node_id: node.id, leased_at: new Date(), heartbeat_at: new Date() })
+
+      const pending = client.get(`/api/v1/node/ranges/${rangeId}/stream`).bearerToken(token)
+      setTimeout(() => Corpus.write({ group: GROUP, start: 1, end: 200 }, { from: 101, to: 200 }, chunk(111), 2), 800)
+      const stream = await pending
+      stream.assertStatus(200)
+      const raw = stream.body() as unknown
+      const body: Buffer = Buffer.isBuffer(raw) ? raw : Buffer.from(stream.text(), 'binary')
+      const frames = []
+      for await (const frame of decodeFrames([body])) frames.push(frame)
+      assert.deepEqual(frames.map((f) => [f.from, f.to]), [[1, 100], [101, 200]])
+      assert.equal(parseOverviews(unzipSync(frames[1].data).toString('utf8')).length, 3)
+      assert.isFalse(await Spool.has(rangeId, { from: 101, to: 200 }))
+    } finally {
+      coordinatorConfig.corpus.dir = saved
+      await rm(corpus, { recursive: true, force: true })
+    }
+  })
+
   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')