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

@@ -544,6 +544,20 @@ releases, a third of them one poster's image dump (27.7 M rows, LOG.md
    (a live reply cut short; the spool doesn't check) is reported after
    the batch before it went out and the range asked for again from that
    chunk: the window stays (the first run lost it, and the open batch).
+   The lease bounds alone don't select a window: the history planner's
+   oldest buckets are priority 0–2 too, so the first run also took 2026
+   history at 0 (25k ranges in 40 groups, fetched live) while 1,739 of
+   the 2016 corpus ranges sat at voyager 2's leftover priorities (done
+   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)
       }

app/services/group_leases.ts

@@ -31,6 +31,11 @@ class GroupLeases {
     return coordinatorConfig.snapshotDir
   }
 
+  /** LEASE_CAPTURE_ONLY: only the corpus windows' ranges (the history's first buckets share their priorities). */
+  private get captureOnly(): string {
+    return coordinatorConfig.leaseCaptureOnly ? 'AND r.capture' : ''
+  }
+
   async lease(nodeId: number): Promise<GroupLease | null> {
     const rules = await RuleSets.current()
     const ttl = coordinatorConfig.leaseTtlSeconds
@@ -43,7 +48,7 @@ class GroupLeases {
       const { rows } = await trx.rawQuery(
         `SELECT g.id, g.name, g.priority, l.node_id AS holder, l.snapshot_next, todo.priority AS todo_priority
          FROM groups g
-         JOIN LATERAL (SELECT min(r.priority) AS priority FROM ranges r WHERE r.group_id = g.id AND r.status = 'planned' AND r.next_id <= r.end_id AND r.priority BETWEEN ? AND ? ${history}) todo ON true
+         JOIN LATERAL (SELECT min(r.priority) AS priority FROM ranges r WHERE r.group_id = g.id AND r.status = 'planned' AND r.next_id <= r.end_id AND r.priority BETWEEN ? AND ? ${history} ${this.captureOnly}) todo ON true
          LEFT JOIN group_leases l ON l.group_id = g.id
          WHERE g.active AND todo.priority IS NOT NULL
            AND (l.node_id IS NULL OR l.heartbeat_at < now() - make_interval(secs => ?))
@@ -62,7 +67,10 @@ class GroupLeases {
           [row.snapshot_next, groupId, row.snapshot_next, row.snapshot_next]
         )
       } else {
-        await trx.rawQuery(`UPDATE ranges SET status = 'planned', node_id = NULL WHERE group_id = ? AND status = 'leased'`, [groupId])
+        await trx.rawQuery(
+          `UPDATE ranges SET status = 'planned', node_id = NULL WHERE group_id = ? AND status = 'leased'`,
+          [groupId]
+        )
       }
       await trx.rawQuery(
         `INSERT INTO group_leases (group_id, node_id, leased_at, heartbeat_at, range_id)
@@ -82,7 +90,8 @@ class GroupLeases {
   /** The group's lease row, when this node holds it. */
   async held(groupId: number, nodeId: number) {
     const lease = await db.from('group_leases').where('group_id', groupId).first()
-    if (!lease || Number(lease.node_id) !== nodeId) throw new LeaseLost(`Group ${groupId} is not held by this node`)
+    if (!lease || Number(lease.node_id) !== nodeId)
+      throw new LeaseLost(`Group ${groupId} is not held by this node`)
     return lease
   }
 
@@ -91,15 +100,21 @@ class GroupLeases {
     await this.held(groupId, nodeId)
     const rules = await RuleSets.current()
     const { leasePriorityMin: min, leasePriorityMax: max } = coordinatorConfig
-    const mine = await db.from('ranges').where('group_id', groupId).where('status', 'leased').where('node_id', nodeId).orderBy('start_id').first()
+    const mine = await db
+      .from('ranges')
+      .where('group_id', groupId)
+      .where('status', 'leased')
+      .where('node_id', nodeId)
+      .orderBy('start_id')
+      .first()
     const row =
       mine ??
       (
         await db.rawQuery(
           `UPDATE ranges r SET status = 'leased', node_id = ?, leased_at = now(), heartbeat_at = now(), attempts = attempts + 1, rule_version = ?
            WHERE r.id = (
-             SELECT id FROM ranges WHERE group_id = ? AND status = 'planned' AND next_id <= end_id AND priority BETWEEN ? AND ?
-             ORDER BY start_id LIMIT 1 FOR UPDATE SKIP LOCKED)
+             SELECT id FROM ranges r WHERE r.group_id = ? AND r.status = 'planned' AND r.next_id <= r.end_id AND r.priority BETWEEN ? AND ? ${this.captureOnly}
+             ORDER BY r.start_id LIMIT 1 FOR UPDATE SKIP LOCKED)
            RETURNING r.id, r.start_id, r.end_id, r.next_id`,
           [nodeId, rules.version, groupId, min, max]
         )
@@ -109,7 +124,14 @@ class GroupLeases {
       .where('group_id', groupId)
       .update({ range_id: row ? Number(row.id) : null, heartbeat_at: new Date() })
     if (!row) return null
-    return { range: { id: Number(row.id), start: Number(row.start_id), end: Number(row.end_id), next: Number(row.next_id) } }
+    return {
+      range: {
+        id: Number(row.id),
+        start: Number(row.start_id),
+        end: Number(row.end_id),
+        next: Number(row.next_id),
+      },
+    }
   }
 
   async heartbeat(groupId: number, nodeId: number) {
@@ -120,14 +142,25 @@ class GroupLeases {
 
   /** A range of the group finished by its holder: counted on the lease (Leases.finish does the range). */
   async rangeDone(groupId: number, headers: number) {
-    await db.rawQuery(`UPDATE group_leases SET ranges_done = ranges_done + 1, headers = headers + ?, range_id = NULL WHERE group_id = ?`, [headers, groupId])
+    await db.rawQuery(
+      `UPDATE group_leases SET ranges_done = ranges_done + 1, headers = headers + ?, range_id = NULL WHERE group_id = ?`,
+      [headers, groupId]
+    )
   }
 
   /** The worker gives the group back (after its snapshot): its range stays at its resume point for the next holder. */
   async release(groupId: number, nodeId: number) {
     await this.held(groupId, nodeId)
-    await db.from('ranges').where('group_id', groupId).where('status', 'leased').where('node_id', nodeId).update({ status: 'planned', node_id: null })
-    await db.from('group_leases').where('group_id', groupId).update({ node_id: null, range_id: null })
+    await db
+      .from('ranges')
+      .where('group_id', groupId)
+      .where('status', 'leased')
+      .where('node_id', nodeId)
+      .update({ status: 'planned', node_id: null })
+    await db
+      .from('group_leases')
+      .where('group_id', groupId)
+      .update({ node_id: null, range_id: null })
   }
 
   private snapshotPath(groupId: number) {
@@ -144,11 +177,17 @@ class GroupLeases {
     const path = this.snapshotPath(groupId)
     await pipeline(body, createWriteStream(`${path}.part`))
     await rename(`${path}.part`, path)
-    await db.from('group_leases').where('group_id', groupId).update({ snapshot_next: next, snapshot_at: new Date(), heartbeat_at: new Date() })
+    await db
+      .from('group_leases')
+      .where('group_id', groupId)
+      .update({ snapshot_next: next, snapshot_at: new Date(), heartbeat_at: new Date() })
   }
 
   /** The snapshot's stream and size, or null when there is none. */
-  async getSnapshot(groupId: number, nodeId: number): Promise<{ stream: Readable; size: number; next: number } | null> {
+  async getSnapshot(
+    groupId: number,
+    nodeId: number
+  ): Promise<{ stream: Readable; size: number; next: number } | null> {
     const lease = await this.held(groupId, nodeId)
     if (lease.snapshot_next === null) return null
     const path = this.snapshotPath(groupId)

app/services/prefetcher.ts

@@ -430,7 +430,11 @@ export class Prefetcher {
       await db
         .from('group_leases')
         .whereNotNull('node_id')
-        .where('heartbeat_at', '>=', new Date(Date.now() - coordinatorConfig.leaseTtlSeconds * 1000))
+        .where(
+          'heartbeat_at',
+          '>=',
+          new Date(Date.now() - coordinatorConfig.leaseTtlSeconds * 1000)
+        )
         .select('group_id')
     ).map((l) => Number(l.group_id))
     // Planned ranges ahead: those started already, then new ones up to SPOOL_AHEAD_PER_LEASE per lease.
@@ -456,8 +460,10 @@ export class Prefetcher {
     const want = Math.max(ahead - Number(ready), short)
     // The catalog at its cap on the NVMe: no history fetched ahead (CatalogSpace).
     const full = await CatalogSpace.full()
-    // Ahead of each held group's cursor: its next ranges in article order, a range of every held group before any
-    // group's second; without held groups (no window worker running) nothing is fetched ahead for workers.
+    // Ahead of each held group's cursor: its next ranges in article order (within what the group leases hand out:
+    // LEASE_MIN/MAX_PRIORITY, LEASE_CAPTURE_ONLY), a range of every held group before any group's second; without
+    // held groups (no window worker running) nothing is fetched ahead for workers.
+    const { leasePriorityMin, leasePriorityMax, leaseCaptureOnly } = coordinatorConfig
     const fresh =
       room && want > 0 && held.length > 0
         ? await rows(db.query())
@@ -465,8 +471,12 @@ export class Prefetcher {
             .where('r.spooled_chunks', 0)
             .whereRaw('r.next_id <= r.end_id')
             .whereIn('r.group_id', held)
+            .whereBetween('r.priority', [leasePriorityMin, leasePriorityMax])
+            .where((q: any) => (leaseCaptureOnly ? q.where('r.capture', true) : q))
             .where((q: any) => (full ? q.where('r.priority', '<', 0) : q))
-            .orderByRaw('row_number() OVER (PARTITION BY r.group_id ORDER BY r.start_id), r.priority, r.start_id')
+            .orderByRaw(
+              'row_number() OVER (PARTITION BY r.group_id ORDER BY r.start_id), r.priority, r.start_id'
+            )
             .limit(want)
         : []
     const targets: Target[] = []
@@ -478,7 +488,8 @@ export class Prefetcher {
       // A chunk already in the corpus isn't spooled: the stream serves it from there (REBUILD.md design 1).
       const corpusRange = row.capture === true ? { group: row.group as string, start, end } : null
       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))) continue
+        if (this.claimed.has(`${row.id}:${span.from}`) || (await Spool.has(Number(row.id), span)))
+          continue
         if (corpusRange && (await Corpus.has(corpusRange, span))) continue
         missing.push(span)
       }

config/coordinator.ts

@@ -33,11 +33,14 @@ export default {
   leaseTtlSeconds: env.get('LEASE_TTL_SECONDS', 15 * 60),
   /**
    * The ranges group leases hand out (REBUILD.md design 1), by priority: the tail is negative, the capture windows
-   * 0–2 (2016, 2021, 2025), the rest of the history above (its position in the group, or 4000+). Bounds so a
-   * deployment can run one window first (homelab 2026-10-26: 0..0, the 2016 window); unbounded by default.
+   * 0–2 (2016, 2021, 2025), the history by its position in the group (0 oldest … 1000 newest: its first buckets
+   * share 0–2 with the windows) or 4000+. Bounds so a deployment can run one window first, with
+   * LEASE_CAPTURE_ONLY so the history's 0 isn't taken along (homelab 2026-10-26: 0..0 and capture only, the 2016
+   * window; LOG.md (118)); unbounded and any range by default.
    */
   leasePriorityMin: env.get('LEASE_MIN_PRIORITY', -1_000_000),
   leasePriorityMax: env.get('LEASE_MAX_PRIORITY', 1_000_000),
+  leaseCaptureOnly: env.get('LEASE_CAPTURE_ONLY', false),
   /** NNTP connections the coordinator fetches headers with: at most (its share, nntpShares), and at least while sharing. */
   nntpConnections: env.get('NNTP_CONNECTIONS', 4),
   nntpMinConnections: env.get('NNTP_MIN_CONNECTIONS', 8),

start/env.ts

@@ -77,6 +77,7 @@ export default await Env.create(new URL('../', import.meta.url), {
   SNAPSHOT_DIR: Env.schema.string.optional(),
   LEASE_MIN_PRIORITY: Env.schema.number.optional(),
   LEASE_MAX_PRIORITY: Env.schema.number.optional(),
+  LEASE_CAPTURE_ONLY: Env.schema.boolean.optional(),
   PENDING_SHARE: Env.schema.number.optional(),
   PENDING_MIN_PROBES: Env.schema.number.optional(),
   PENDING_MAX_NAMED: Env.schema.number.optional(),

tests/functional/group_leases.spec.ts

@@ -133,6 +133,32 @@ test.group('Group leases', (g) => {
     assert.equal((await GroupLeases.next(a, n1))?.range.id, r4)
   })
 
+  test('capture only: the history ranges sharing a window priority are neither leased nor handed out', async ({
+    assert,
+  }) => {
+    // Group e: history at priority 0 only (the planner's oldest bucket); group f: a 2016 window range (capture)
+    // between two history ranges at 0.
+    const e = await group('alt.binaries.gl.e', 1)
+    await range(e, 1, 0)
+    const f = await group('alt.binaries.gl.f', 2)
+    await range(f, 1, 0)
+    const f2 = await range(f, 101, 0, { capture: true })
+    await range(f, 201, 0)
+    const n1 = await node('gl-8')
+    const was = coordinatorConfig.leaseCaptureOnly
+    coordinatorConfig.leaseCaptureOnly = true
+    try {
+      assert.equal((await GroupLeases.lease(n1))?.group.id, f, 'the group with a capture range, not the older one')
+      assert.equal((await GroupLeases.next(f, n1))?.range.id, f2)
+      await db.from('ranges').where('id', f2).update({ status: 'done', next_id: 201 })
+      assert.isNull(await GroupLeases.next(f, n1), 'the history range after the window is not handed out')
+      assert.isNull(await GroupLeases.lease(await node('gl-9')))
+    } finally {
+      coordinatorConfig.leaseCaptureOnly = was
+    }
+    assert.equal((await GroupLeases.lease(await node('gl-10')))?.group.id, e, 'any range again without the flag')
+  })
+
   test('a snapshot older than a done range makes that range be read again from its start', async ({ assert }) => {
     const a = await group('alt.binaries.gl.d', 1)
     const r1 = await range(a, 1, 0)

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')