LEASE_CAPTURE_ONLY: group leases of the corpus windows' ranges only

The lease bounds alone don't select a window: the history planner's oldest
buckets are priority 0-2 as well, so the window worker at 0..0 also took 2026
history at 0 (25k ranges in 40 groups, fetched live) while 1,739 ranges of the
2016 corpus window sat at voyager 2's leftover priorities (done when the window
was marked, planned again by the reset). With LEASE_CAPTURE_ONLY the lease, next
and the prefetcher's run-ahead take capture ranges only; the run-ahead now also
keeps to the lease bounds. Test: a group with history at 0 only isn't leased,
the history range after the window isn't handed out.

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

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