An "ancient" range a group: articles 1 … low watermark − 1, never filled

Every group is tiled from article 1 now: one range with status 'ancient'
covers what had expired on the server before the group was planned.
Nothing selects it (leases, prefetcher, corpus fill, tail merger all take
'planned'), the status panel reports it apart, the search's history stats
leave it out. Migration 6700 adds it for the groups planned before;
Planner.historyRows (now a pure function, unit-tested) for new groups.
The status bands are ordered by an explicit key: the ancient row at
priority 0 tied "history 0-999" with "window 0".

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

.gitea/workflows/deploy.yaml

@@ -59,35 +59,28 @@ jobs:
           node-version-file: .nvmrc
           cache: npm
       - run: npm ci
+      - name: Rust toolchain
+        run: |
+          command -v cargo || curl -sSf https://sh.rustup.rs | sh -s -- -y --profile minimal
+          echo "$HOME/.cargo/bin" >> "$GITHUB_PATH"
+      # The edge worker (edge/, Rust): built first, since the e2e suite runs edge/target/release/edge-node against
+      # the node API; then its own tests, and the Rust and TypeScript RangeIndexers compared on the committed
+      # captures (edge/fixtures).
+      - name: Build the edge worker
+        run: cargo build --release --manifest-path edge/Cargo.toml --bin edge-node
       - name: Type check
         run: npx tsc --noEmit
       - name: Migrate
         run: node ace migration:run --force
       - name: Test
         run: node ace test
-
-  # The edge worker (edge/, Rust): its tests, and the Rust and TypeScript
-  # RangeIndexers compared on the committed captures (edge/fixtures).
-  edge:
-    runs-on: ubuntu-latest
-    steps:
-      - uses: actions/checkout@v4
-      - uses: actions/setup-node@v4
-        with:
-          node-version-file: .nvmrc
-          cache: npm
-      - run: npm ci
-      - name: Rust toolchain
-        run: |
-          command -v cargo || curl -sSf https://sh.rustup.rs | sh -s -- -y --profile minimal
-          echo "$HOME/.cargo/bin" >> "$GITHUB_PATH"
       - name: cargo test
         run: cargo test --release --manifest-path edge/Cargo.toml
       - name: The Rust indexer over the fixtures
         run: edge/bench-fixtures.sh
 
   deploy:
-    needs: [test, edge]
+    needs: [test]
     if: github.ref == 'refs/heads/rebuild'
     runs-on: ubuntu-latest
     steps:

REBUILD.md

@@ -60,6 +60,13 @@ most workers on batcave, a few on poweredge-kantoor with
 `edge:node --max-mbit` (e.g. 50 of its 160 Mbit/s ≈ 100–125 k articles/s;
 the cap is shared by the node's processes).
 
+Every group is tiled from article 1 (2026-10-28): one "ancient" range,
+status `ancient`, covers 1 … low watermark − 1, what had expired on the
+server before the group was planned. Never leased, fetched or filled —
+nothing selects it — and the status panel reports it apart ("Ancient: …
+articles below the low watermark"); migration 6700 added it for the
+groups planned before, `Planner.historyRows` for groups planned since.
+
 Also built (2026-10-09): the live tail (new articles planned every 2
 minutes from one LIST ACTIVE per name prefix, at priority −1: leased and
 fetched before history; since 2026-10-27 a round grows the group's open
@@ -573,6 +580,40 @@ releases, a third of them one poster's image dump (27.7 M rows, LOG.md
    open batch before the group goes back, like a stop does, so the
    window's snapshot survives it; only a batch the coordinator refused
    (lease lost, too large) still loses the window.
+   **Two lanes a group (2026-10-27; migration 6600, LOG.md (119)).** A
+   group's ranges in article order through one window meant a thread
+   that leased a group for its tail range first waded through the
+   group's history (older articles come first), and a held history
+   group's tail waited behind days of its ranges; and the bounds that ran
+   the 2016 window alone (0..0, capture only) shut the tail out entirely.
+   Now `group_leases` has a row per group and *lane*: the `history` lane
+   works the ranges the bounds hand out (the windows, the history), the
+   `tail` lane the tail planner's (priority at or below the tail's,
+   whatever the bounds), each with its own holder, window, snapshot
+   (`<group>-tail.gz`) and takeover redo (the lane's ranges only, and
+   only those the window did after the snapshot: leased, done since it,
+   or read past it). The edge puts `--tail-threads` of its threads on the
+   tail lane (EDGE_TAIL_THREADS; 1 of 2 or more by default, which is
+   what homelab's 16 run with), the rest on history; the prefetcher runs ahead of every held
+   cursor. The seam a group is the same one as before, once: where the
+   history lane reaches the first range the tail lane did, it is caught
+   up. **The lease order** within a lane and priority is a mix (the
+   user, 2026-10-27): one lease starts from the group with the most
+   planned articles, the next from the longest unheld one (never held
+   first), alternating on a sequence (`group_lease_turns`), so a big
+   group snapshotted at a stop doesn't wait behind hundreds of small
+   never-held groups and a quiet group's tail is visited all the same.
+   **`coordinator:status`** (`app/services/status.ts`, `--json`) prints
+   the catalog's room against `CATALOG_MAX_NVME_GB`, the ranges by band
+   (tail, raised, each capture window, the history by thousand), the
+   last hour's indexed and stored articles, the spool, the tail backlog,
+   the held groups with their snapshots' age and size, the snapshot
+   files, and the phase-2 backlogs (provisional releases, probe
+   candidates, the pending ladder, the judge's dirty marks). **End to
+   end** (`tests/e2e`, suite `e2e`, CI builds the binary first): the Rust
+   worker against the real node API, database, spool and corpus — the
+   corpus-fill race, a failed stream and the restart that restores the
+   snapshot, a restart whose snapshot file is gone, a takeover.
 
 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

adonisrc.ts

@@ -91,6 +91,12 @@ export default defineConfig({
         name: 'functional',
         timeout: 30000,
       },
+      // The Rust edge worker against the real node API (tests/e2e): needs edge/target/release/edge-node.
+      {
+        files: ['tests/e2e/**/*.spec.{ts,js}'],
+        name: 'e2e',
+        timeout: 120000,
+      },
     ],
     forceExit: false,
   },

app/controllers/node_controller.ts

@@ -3,18 +3,13 @@ import { Readable } from 'node:stream'
 import { setTimeout as sleep } from 'node:timers/promises'
 import db from '@adonisjs/lucid/services/db'
 import Leases, { LeaseLost } from '#services/leases'
-import GroupLeases from '#services/group_leases'
+import GroupLeases, { LANES } from '#services/group_leases'
 import BatchStore, { StaleBatch } from '#services/batch_store'
 import Spool, { type ChunkSpan } from '#services/spool'
 import Corpus from '#services/corpus'
-import { KEEPALIVE, encodeFrame, fromWire, type WireBatch } from '#lib/edge/index'
+import coordinatorConfig from '#config/coordinator'
+import { KEEPALIVE, encodeFrame, fromWire, type Lane, type WireBatch } from '#lib/edge/index'
 
-/**
- * How long the stream waits for a chunk the prefetcher hasn't spooled yet. Long: a chunk that keeps
- * coming in isn't given up (only a stalled one is), and some groups' chunks take 20+ minutes
- * (a.b.0day.stuffz, 17 MB in 22 min; LOG.md 2026-10-10 (3)).
- */
-const CHUNK_WAIT_MS = 60 * 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
 
@@ -35,16 +30,29 @@ export default class NodeController {
 
   // Group leases (REBUILD.md, "Voyager 3: window indexing", 1 and 5): a worker thread holds a group and takes its
   // ranges in article order; the window's snapshot goes through here too, so a worker needs no disk of its own.
+  // `?lane=tail` is the group's tail cursor (its own holder, window and snapshot); history by default.
+
+  /** The lane the request names (`?lane=`), or null after a 400 for one that isn't a lane. */
+  private lane(ctx: HttpContext): Lane | null {
+    const lane = ctx.request.qs().lane ?? 'history'
+    if (LANES.includes(lane)) return lane
+    ctx.response.badRequest({ error: `lane: ${LANES.join(' or ')}, not "${lane}"` })
+    return null
+  }
 
   async leaseGroup(ctx: HttpContext) {
-    const lease = await GroupLeases.lease(this.node(ctx))
+    const lane = this.lane(ctx)
+    if (!lane) return
+    const lease = await GroupLeases.lease(this.node(ctx), lane)
     if (!lease) return ctx.response.noContent()
     return lease
   }
 
   async nextRange(ctx: HttpContext) {
+    const lane = this.lane(ctx)
+    if (!lane) return
     try {
-      const next = await GroupLeases.next(Number(ctx.params.id), this.node(ctx))
+      const next = await GroupLeases.next(Number(ctx.params.id), this.node(ctx), lane)
       if (!next) return ctx.response.noContent()
       return next
     } catch (error) {
@@ -54,11 +62,15 @@ export default class NodeController {
   }
 
   async groupHeartbeat(ctx: HttpContext) {
-    return this.lifecycle(ctx, (groupId, nodeId) => GroupLeases.heartbeat(groupId, nodeId))
+    const lane = this.lane(ctx)
+    if (!lane) return
+    return this.lifecycle(ctx, (groupId, nodeId) => GroupLeases.heartbeat(groupId, nodeId, lane))
   }
 
   async releaseGroup(ctx: HttpContext) {
-    return this.lifecycle(ctx, (groupId, nodeId) => GroupLeases.release(groupId, nodeId))
+    const lane = this.lane(ctx)
+    if (!lane) return
+    return this.lifecycle(ctx, (groupId, nodeId) => GroupLeases.release(groupId, nodeId, lane))
   }
 
   /**
@@ -66,16 +78,20 @@ export default class NodeController {
    * (gzip; the body parser leaves an octet-stream alone, so the request is still readable here).
    */
   async putSnapshot(ctx: HttpContext) {
+    const lane = this.lane(ctx)
+    if (!lane) return
     const next = Number(ctx.request.qs().next)
     if (!Number.isFinite(next) || next <= 0) return ctx.response.badRequest({ error: 'next (the snapshot\'s article) is needed' })
     const raw = ctx.request.raw()
     const body = raw !== null ? Readable.from([Buffer.from(raw, 'binary')]) : ctx.request.request
-    return this.lifecycle(ctx, (groupId, nodeId) => GroupLeases.putSnapshot(groupId, nodeId, next, body))
+    return this.lifecycle(ctx, (groupId, nodeId) => GroupLeases.putSnapshot(groupId, nodeId, next, body, lane))
   }
 
   async getSnapshot(ctx: HttpContext) {
+    const lane = this.lane(ctx)
+    if (!lane) return
     try {
-      const snapshot = await GroupLeases.getSnapshot(Number(ctx.params.id), this.node(ctx))
+      const snapshot = await GroupLeases.getSnapshot(Number(ctx.params.id), this.node(ctx), lane)
       if (!snapshot) return ctx.response.notFound({ error: 'no snapshot' })
       ctx.response.header('Content-Type', 'application/octet-stream')
       ctx.response.header('Content-Length', String(snapshot.size))
@@ -107,7 +123,7 @@ export default class NodeController {
       corpusRange && (await Corpus.has(corpusRange, span)) ? Corpus.read(corpusRange, span) : null
     async function* frames() {
       for (const span of spans) {
-        const waitUntil = Date.now() + CHUNK_WAIT_MS
+        const waitUntil = Date.now() + coordinatorConfig.streamChunkWaitMs
         let quietSince = Date.now()
         let data = await stored(span)
         while (!data) {
@@ -152,7 +168,7 @@ export default class NodeController {
     return this.lifecycle(ctx, async (rangeId, nodeId) => {
       const range = await Leases.leased(rangeId, nodeId)
       await Leases.finish(rangeId, nodeId)
-      await GroupLeases.rangeDone(Number(range.group_id), Number(range.headers))
+      await GroupLeases.rangeDone(Number(range.group_id), Number(range.headers), GroupLeases.laneOf(Number(range.priority)))
     })
   }
 

app/services/group_leases.ts

@@ -8,120 +8,179 @@ import coordinatorConfig from '#config/coordinator'
 import RuleSets from '#services/rule_sets'
 import CatalogSpace from '#services/catalog_space'
 import { LeaseLost } from '#services/leases'
-import type { GroupLease, NextRange } from '#lib/edge/index'
+import type { GroupLease, Lane, NextRange } from '#lib/edge/index'
+
+export const LANES: Lane[] = ['history', 'tail']
 
 /**
  * Group leases (REBUILD.md, "Voyager 3: window indexing", 1 and 5): a worker thread holds a group and asks for its
  * next range each time it finished one, so a group's ranges are indexed in article order through one carried
- * window. The coordinator keeps one row a group (group_leases): the holder, its heartbeat, the range it works and
+ * window. The coordinator keeps one row a group and lane (group_leases): the holder, its heartbeat, the range it
- * the snapshot its window stands at.
+ * works and the snapshot its window stands at.
  *
- * - lease: the free group with the most urgent planned range (the ranges' priority: the tail before history, then
+ * - Two lanes (2026-10-27): the `history` lane works the ranges the bounds hand out (LEASE_MIN/MAX_PRIORITY,
- *   the windows), the longest unheld first; a holder quiet for LEASE_TTL_SECONDS is taken over. Taking a group
+ *   LEASE_CAPTURE_ONLY: the windows, the history), the `tail` lane the tail planner's ranges (priority at or below
- *   resets what the window's snapshot doesn't cover: the ranges from the snapshot's article on are planned again
+ *   the tail's). Each lane has its own holder, window and snapshot, so a thread on the tail never wades through a
- *   (the batch store's keys make the redo idempotent); without a snapshot the window is lost and the group
+ *   group's history first (a group's ranges come in article order, and a held group's history is days of ranges), and
- *   continues at its resume point (one seam).
+ *   the tail of a group whose history is held isn't stuck behind it. The one seam a group is where the history lane
- * - next: the group's lowest planned range (or the one this node already has), leased to the node; none when the
+ *   reaches the tail lane's first range (its ranges are done: the history lane is caught up).
- *   group is caught up: the worker snapshots and gives the group back, and gets it again when the tail planned
+ * - lease: the free group with the most urgent planned range of the lane (the ranges' priority), and among those a
- *   more.
+ *   mix: the group with the most planned articles and the longest unheld one (never held first) take turns, so a
- * - snapshot: the serialized window, kept as a file on bulk (SNAPSHOT_DIR), given to the next holder.
+ *   big group snapshotted at a stop doesn't wait behind hundreds of small never-held ones, and a quiet group's tail
+ *   is visited all the same. A holder quiet for LEASE_TTL_SECONDS is taken over. Taking a group resets what the
+ *   lane's snapshot doesn't cover: the lane's ranges from the snapshot's article on that the window did after it
+ *   (leased, done since the snapshot, or read past it) are planned again (the batch store's keys make the redo
+ *   idempotent); without a snapshot the window is lost and the group continues at its resume point (one seam).
+ * - next: the group's lowest planned range of the lane (or the one this node already has), leased to the node; none
+ *   when the lane is caught up: the worker snapshots and gives the group back, and gets it again when the tail
+ *   planned more.
+ * - snapshot: the serialized window, kept as a file on bulk (SNAPSHOT_DIR; `<group>.gz`, the tail's
+ *   `<group>-tail.gz`), given to the next holder of the lane.
  */
 class GroupLeases {
   get dir(): string {
     return coordinatorConfig.snapshotDir
   }
 
-  /** LEASE_CAPTURE_ONLY: only the corpus windows' ranges (the history's first buckets share their priorities). */
+  /**
-  private get captureOnly(): string {
+   * The lane's ranges, as a condition on `r`: the tail lane's are at or below the tail's priority, the history lane's
-    return coordinatorConfig.leaseCaptureOnly ? 'AND r.capture' : ''
+   * above it — within the lease bounds when `bounds` (what is handed out), or all of them (what a lane's window did).
+   */
+  private laneFilter(lane: Lane, bounds: boolean): { sql: string; bindings: number[] } {
+    const tail = coordinatorConfig.tail.priority
+    if (lane === 'tail') return { sql: 'r.priority <= ?', bindings: [tail] }
+    if (!bounds) return { sql: 'r.priority > ?', bindings: [tail] }
+    const { leasePriorityMin: min, leasePriorityMax: max, leaseCaptureOnly } = coordinatorConfig
+    return {
+      sql: `r.priority > ? AND r.priority BETWEEN ? AND ?${leaseCaptureOnly ? ' AND r.capture' : ''}`,
+      bindings: [tail, min, max],
+    }
   }
 
-  async lease(nodeId: number): Promise<GroupLease | null> {
+  async lease(nodeId: number, lane: Lane = 'history'): Promise<GroupLease | null> {
+    // The catalog at its cap on the NVMe: the tail only (CatalogSpace), as the range leases did.
+    if (lane === 'history' && (await CatalogSpace.full())) return null
     const rules = await RuleSets.current()
     const ttl = coordinatorConfig.leaseTtlSeconds
-    const { leasePriorityMin: min, leasePriorityMax: max } = coordinatorConfig
+    const todo = this.laneFilter(lane, true)
-    // The catalog at its cap on the NVMe: the tail only (CatalogSpace), as the range leases did.
+    const all = this.laneFilter(lane, false)
-    const history = (await CatalogSpace.full()) ? 'AND r.priority < 0' : ''
     return db.transaction(async (trx) => {
       // One lease at a time: the choice reads several tables.
       await trx.rawQuery(`SELECT pg_advisory_xact_lock(hashtextextended('group-lease', 0))`)
+      // The free groups with planned work in the lane (a range leased by a holder gone quiet counts: a group whose
+      // last range that holder took would never be taken over), by the most urgent range's priority; within a priority the
+      // most planned articles and the longest unheld alternate: one lease starts from the biggest, the next from
+      // the longest unheld (the sequence counts the leases; ranked afresh each time, "biggest first" would always
+      // win).
+      const { rows: turns } = await trx.rawQuery(`SELECT nextval('group_lease_turns') % 2 AS odd`)
+      const [first, second] = Number(turns[0].odd) === 1 ? [1, 0] : [0, 1]
       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
+        `WITH todo AS (
-         FROM groups g
+           SELECT r.group_id, min(r.priority) AS priority, sum(r.end_id - r.next_id + 1) AS articles
-         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
+           FROM ranges r
-         LEFT JOIN group_leases l ON l.group_id = g.id
+           WHERE (r.status = 'planned' OR (r.status = 'leased' AND r.heartbeat_at < now() - make_interval(secs => ?)))
-         WHERE g.active AND todo.priority IS NOT NULL
+             AND r.next_id <= r.end_id AND ${todo.sql}
-           AND (l.node_id IS NULL OR l.heartbeat_at < now() - make_interval(secs => ?))
+           GROUP BY r.group_id
-         ORDER BY todo.priority, l.heartbeat_at NULLS FIRST, g.priority
+         ), free AS (
-         LIMIT 1`,
+           SELECT g.id, g.name, g.priority AS group_priority, l.snapshot_next, l.snapshot_at, l.heartbeat_at,
-        [min, max, ttl]
+                  todo.priority, todo.articles
+           FROM todo JOIN groups g ON g.id = todo.group_id AND g.active
+           LEFT JOIN group_leases l ON l.group_id = g.id AND l.lane = ?
+           WHERE l.node_id IS NULL OR l.heartbeat_at < now() - make_interval(secs => ?)
+         ), turns AS (
+           SELECT free.*,
+                  least(2 * row_number() OVER (PARTITION BY priority ORDER BY articles DESC, group_priority, id) - ?,
+                        2 * row_number() OVER (PARTITION BY priority ORDER BY heartbeat_at NULLS FIRST, group_priority, id) - ?) AS turn
+           FROM free
+         )
+         SELECT * FROM turns ORDER BY priority, turn, group_priority, id LIMIT 1`,
+        [ttl, ...todo.bindings, lane, ttl, first, second]
       )
       const row = rows[0]
       if (!row) return null
       const groupId = Number(row.id)
-      // What the snapshot doesn't cover is read again: the ranges from its article on, the one holding it from there.
+      // What the snapshot doesn't cover is read again: the lane's ranges from its article on that the window did
+      // after it (a range done before the snapshot was taken — by the old range leases, say — stays done).
       if (row.snapshot_next !== null) {
         await trx.rawQuery(
-          `UPDATE ranges SET status = 'planned', node_id = NULL, next_id = greatest(start_id, ?::bigint)
+          `UPDATE ranges r SET status = 'planned', node_id = NULL, next_id = greatest(r.start_id, ?::bigint)
-           WHERE group_id = ? AND end_id >= ?::bigint AND (status IN ('leased', 'done') OR next_id > ?::bigint)`,
+           WHERE r.group_id = ? AND r.end_id >= ?::bigint AND ${all.sql}
-          [row.snapshot_next, groupId, row.snapshot_next, row.snapshot_next]
+             AND (r.status = 'leased' OR (r.status = 'done' AND (r.done_at IS NULL OR r.done_at >= ?)) OR (r.status = 'planned' AND r.next_id > ?::bigint))`,
+          [
+            row.snapshot_next,
+            groupId,
+            row.snapshot_next,
+            ...all.bindings,
+            row.snapshot_at ?? new Date(0),
+            row.snapshot_next,
+          ]
         )
       } else {
         await trx.rawQuery(
-          `UPDATE ranges SET status = 'planned', node_id = NULL WHERE group_id = ? AND status = 'leased'`,
+          `UPDATE ranges r SET status = 'planned', node_id = NULL WHERE r.group_id = ? AND r.status = 'leased' AND ${all.sql}`,
-          [groupId]
+          [groupId, ...all.bindings]
         )
       }
       await trx.rawQuery(
-        `INSERT INTO group_leases (group_id, node_id, leased_at, heartbeat_at, range_id)
+        `INSERT INTO group_leases (group_id, lane, node_id, leased_at, heartbeat_at, range_id)
-         VALUES (?, ?, now(), now(), NULL)
+         VALUES (?, ?, ?, now(), now(), NULL)
-         ON CONFLICT (group_id) DO UPDATE SET node_id = excluded.node_id, leased_at = now(), heartbeat_at = now(), range_id = NULL`,
+         ON CONFLICT (group_id, lane) DO UPDATE SET node_id = excluded.node_id, leased_at = now(), heartbeat_at = now(), range_id = NULL`,
-        [groupId, nodeId]
+        [groupId, lane, nodeId]
       )
       await trx.from('nodes').where('id', nodeId).update({ last_seen_at: new Date() })
       return {
-        group: { id: groupId, name: row.name, priority: Number(row.priority) },
+        group: { id: groupId, name: row.name, priority: Number(row.group_priority) },
+        lane,
         rules,
         snapshot: row.snapshot_next !== null ? { next: Number(row.snapshot_next) } : null,
       }
     })
   }
 
-  /** The group's lease row, when this node holds it. */
+  /** The group's lease row of the lane, when this node holds it. */
-  async held(groupId: number, nodeId: number) {
+  async held(groupId: number, nodeId: number, lane: Lane = 'history') {
-    const lease = await db.from('group_leases').where('group_id', groupId).first()
+    const lease = await db
+      .from('group_leases')
+      .where('group_id', groupId)
+      .where('lane', lane)
+      .first()
     if (!lease || Number(lease.node_id) !== nodeId)
-      throw new LeaseLost(`Group ${groupId} is not held by this node`)
+      throw new LeaseLost(`Group ${groupId} (${lane}) is not held by this node`)
     return lease
   }
 
-  /** The group's next range in article order (or the one this node has), leased to the node; null when caught up. */
+  /**
-  async next(groupId: number, nodeId: number): Promise<NextRange | null> {
+   * The group's next range of the lane in article order (or the one this node has), leased to the node; null when
-    await this.held(groupId, nodeId)
+   * the lane is caught up.
+   */
+  async next(groupId: number, nodeId: number, lane: Lane = 'history'): Promise<NextRange | null> {
+    await this.held(groupId, nodeId, lane)
     const rules = await RuleSets.current()
-    const { leasePriorityMin: min, leasePriorityMax: max } = coordinatorConfig
+    const todo = this.laneFilter(lane, true)
-    const mine = await db
+    const all = this.laneFilter(lane, false)
-      .from('ranges')
+    const mine = (
-      .where('group_id', groupId)
+      await db.rawQuery(
-      .where('status', 'leased')
+        `SELECT r.id, r.start_id, r.end_id, r.next_id FROM ranges r
-      .where('node_id', nodeId)
+         WHERE r.group_id = ? AND r.status = 'leased' AND r.node_id = ? AND ${all.sql} ORDER BY r.start_id LIMIT 1`,
-      .orderBy('start_id')
+        [groupId, nodeId, ...all.bindings]
-      .first()
+      )
+    ).rows[0]
     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 r WHERE r.group_id = ? AND r.status = 'planned' AND r.next_id <= r.end_id AND r.priority BETWEEN ? AND ? ${this.captureOnly}
+             SELECT r.id FROM ranges r WHERE r.group_id = ? AND r.status = 'planned' AND r.next_id <= r.end_id AND ${todo.sql}
              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]
+          [nodeId, rules.version, groupId, ...todo.bindings]
         )
       ).rows[0]
     await db
       .from('group_leases')
       .where('group_id', groupId)
+      .where('lane', lane)
       .update({ range_id: row ? Number(row.id) : null, heartbeat_at: new Date() })
     if (!row) return null
     return {
@@ -134,63 +193,81 @@ class GroupLeases {
     }
   }
 
-  async heartbeat(groupId: number, nodeId: number) {
+  async heartbeat(groupId: number, nodeId: number, lane: Lane = 'history') {
-    await this.held(groupId, nodeId)
+    await this.held(groupId, nodeId, lane)
-    await db.from('group_leases').where('group_id', groupId).update({ heartbeat_at: new Date() })
+    await db
+      .from('group_leases')
+      .where('group_id', groupId)
+      .where('lane', lane)
+      .update({ heartbeat_at: new Date() })
     await db.from('nodes').where('id', nodeId).update({ last_seen_at: new Date() })
   }
 
-  /** A range of the group finished by its holder: counted on the lease (Leases.finish does the range). */
+  /** The lane a range belongs to, by its priority. */
-  async rangeDone(groupId: number, headers: number) {
+  laneOf(priority: number): Lane {
+    return priority <= coordinatorConfig.tail.priority ? 'tail' : 'history'
+  }
+
+  /** A range of the group finished by its holder: counted on the lane's lease (Leases.finish does the range). */
+  async rangeDone(groupId: number, headers: number, lane: Lane = 'history') {
     await db.rawQuery(
-      `UPDATE group_leases SET ranges_done = ranges_done + 1, headers = headers + ?, range_id = NULL WHERE group_id = ?`,
+      `UPDATE group_leases SET ranges_done = ranges_done + 1, headers = headers + ?, range_id = NULL WHERE group_id = ? AND lane = ?`,
-      [headers, groupId]
+      [headers, groupId, lane]
     )
   }
 
   /** 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) {
+  async release(groupId: number, nodeId: number, lane: Lane = 'history') {
-    await this.held(groupId, nodeId)
+    await this.held(groupId, nodeId, lane)
-    await db
+    const all = this.laneFilter(lane, false)
-      .from('ranges')
+    await db.rawQuery(
-      .where('group_id', groupId)
+      `UPDATE ranges r SET status = 'planned', node_id = NULL
-      .where('status', 'leased')
+       WHERE r.group_id = ? AND r.status = 'leased' AND r.node_id = ? AND ${all.sql}`,
-      .where('node_id', nodeId)
+      [groupId, nodeId, ...all.bindings]
-      .update({ status: 'planned', node_id: null })
+    )
     await db
       .from('group_leases')
       .where('group_id', groupId)
+      .where('lane', lane)
       .update({ node_id: null, range_id: null })
   }
 
-  private snapshotPath(groupId: number) {
+  snapshotPath(groupId: number, lane: Lane = 'history') {
-    return join(this.dir, `${groupId}.gz`)
+    return join(this.dir, lane === 'tail' ? `${groupId}-tail.gz` : `${groupId}.gz`)
   }
 
   /**
    * The window's snapshot standing at `next` (the next article to index), streamed to disk (a busy group's is
    * gigabytes) and written whole before it replaces the old one.
    */
-  async putSnapshot(groupId: number, nodeId: number, next: number, body: Readable) {
+  async putSnapshot(
-    await this.held(groupId, nodeId)
+    groupId: number,
+    nodeId: number,
+    next: number,
+    body: Readable,
+    lane: Lane = 'history'
+  ) {
+    await this.held(groupId, nodeId, lane)
     await mkdir(this.dir, { recursive: true })
-    const path = this.snapshotPath(groupId)
+    const path = this.snapshotPath(groupId, lane)
     await pipeline(body, createWriteStream(`${path}.part`))
     await rename(`${path}.part`, path)
     await db
       .from('group_leases')
       .where('group_id', groupId)
+      .where('lane', lane)
       .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
+    nodeId: number,
+    lane: Lane = 'history'
   ): Promise<{ stream: Readable; size: number; next: number } | null> {
-    const lease = await this.held(groupId, nodeId)
+    const lease = await this.held(groupId, nodeId, lane)
     if (lease.snapshot_next === null) return null
-    const path = this.snapshotPath(groupId)
+    const path = this.snapshotPath(groupId, lane)
     try {
       const { size } = await stat(path)
       return { stream: createReadStream(path), size, next: Number(lease.snapshot_next) }

app/services/planner.ts

@@ -35,20 +35,8 @@ class Planner {
 
   async planGroup(group: { id: number; name: string }): Promise<number> {
     const info = await NntpPool.use((client) => client.group(group.name))
-    const size = coordinatorConfig.rangeArticles
     const last = await db.from('ranges').where('group_id', group.id).max('end_id as end').first()
-    const from = Math.max(info.low, Number(last?.end ?? 0) + 1)
+    const rows = historyRows(group.id, info, Number(last?.end ?? 0), coordinatorConfig.rangeArticles)
-    const span = Math.max(1, info.high - info.low)
-    const rows = []
-    for (let start = from; start <= info.high; start += size) {
-      rows.push({
-        group_id: group.id,
-        start_id: start,
-        end_id: Math.min(start + size - 1, info.high),
-        next_id: start,
-        priority: Math.floor((1000 * (start - info.low)) / span),
-      })
-    }
     for (let i = 0; i < rows.length; i += 2000) await db.table('ranges').multiInsert(rows.slice(i, i + 2000))
     await db.from('groups').where('id', group.id).update({ low: info.low, high: info.high, planned_at: new Date() })
     return rows.length
@@ -270,4 +258,34 @@ export function tailMerges(
   return merges
 }
 
+/**
+ * The rows a group's history is cut into, from `lastEnd + 1` (0: the group has no ranges yet) up to the high
+ * watermark: RANGE_ARTICLES slices at the position priority. A group planned for the first time also gets one
+ * "ancient" range, articles 1 … low − 1: what had expired on the server before it was planned. Status
+ * 'ancient', never leased or fetched; it keeps the tiling complete from article 1 (migration 6700 added it for
+ * the groups planned before 2026-10-28).
+ */
+export function historyRows(
+  groupId: number,
+  info: { low: number; high: number },
+  lastEnd: number,
+  size: number
+): Record<string, unknown>[] {
+  const rows: Record<string, unknown>[] = []
+  if (lastEnd === 0 && info.low > 1)
+    rows.push({ group_id: groupId, start_id: 1, end_id: info.low - 1, next_id: 1, priority: 0, status: 'ancient' })
+  const from = Math.max(info.low, lastEnd + 1)
+  const span = Math.max(1, info.high - info.low)
+  for (let start = from; start <= info.high; start += size) {
+    rows.push({
+      group_id: groupId,
+      start_id: start,
+      end_id: Math.min(start + size - 1, info.high),
+      next_id: start,
+      priority: Math.floor((1000 * (start - info.low)) / span),
+    })
+  }
+  return rows
+}
+
 export default new Planner()

app/services/prefetcher.ts

@@ -425,8 +425,9 @@ export class Prefetcher {
       )
     // 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')
-    // The groups workers hold (group_leases; REBUILD.md design 1): the prefetcher runs ahead of their cursors.
+    // The groups workers hold (group_leases; REBUILD.md design 1), by lane: the prefetcher runs ahead of their
-    const held: number[] = (
+    // cursors (a group's history and tail lanes are two cursors).
+    const held: { group: number; lane: string }[] = (
       await db
         .from('group_leases')
         .whereNotNull('node_id')
@@ -435,8 +436,8 @@ export class Prefetcher {
           '>=',
           new Date(Date.now() - coordinatorConfig.leaseTtlSeconds * 1000)
         )
-        .select('group_id')
+        .select('group_id', 'lane')
-    ).map((l) => Number(l.group_id))
+    ).map((l) => ({ group: Number(l.group_id), lane: l.lane }))
     // Planned ranges ahead: those started already, then new ones up to SPOOL_AHEAD_PER_LEASE per lease.
     const ahead = Math.max(
       coordinatorConfig.spool.minAhead,
@@ -460,22 +461,36 @@ 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 (within what the group leases hand out:
+    // Ahead of each held cursor: the lane's next ranges in article order (the history lane's within what the group
-    // LEASE_MIN/MAX_PRIORITY, LEASE_CAPTURE_ONLY), a range of every held group before any group's second; without
+    // leases hand out: LEASE_MIN/MAX_PRIORITY, LEASE_CAPTURE_ONLY, and none at the catalog's cap; the tail lane's
-    // held groups (no window worker running) nothing is fetched ahead for workers.
+    // at the tail's priority), a range of every held cursor before any cursor's second; without held groups (no
+    // window worker running) nothing is fetched ahead for workers.
     const { leasePriorityMin, leasePriorityMax, leaseCaptureOnly } = coordinatorConfig
+    const tailPriority = coordinatorConfig.tail.priority
+    const heldHistory = full ? [] : held.filter((h) => h.lane === 'history').map((h) => h.group)
+    const heldTail = held.filter((h) => h.lane === 'tail').map((h) => h.group)
     const fresh =
-      room && want > 0 && held.length > 0
+      room && want > 0 && heldHistory.length + heldTail.length > 0
         ? await rows(db.query())
             .where('r.status', 'planned')
             .where('r.spooled_chunks', 0)
             .whereRaw('r.next_id <= r.end_id')
-            .whereIn('r.group_id', held)
+            .where((q: any) =>
+              q
+                .where((h: any) =>
+                  h
+                    .whereIn('r.group_id', heldHistory)
+                    .where('r.priority', '>', tailPriority)
                     .whereBetween('r.priority', [leasePriorityMin, leasePriorityMax])
-            .where((q: any) => (leaseCaptureOnly ? q.where('r.capture', true) : q))
+                    .where((c: any) => (leaseCaptureOnly ? c.where('r.capture', true) : c))
-            .where((q: any) => (full ? q.where('r.priority', '<', 0) : q))
+                )
+                .orWhere((t: any) =>
+                  t.whereIn('r.group_id', heldTail).where('r.priority', '<=', tailPriority)
+                )
+            )
             .orderByRaw(
-              'row_number() OVER (PARTITION BY r.group_id ORDER BY r.start_id), r.priority, r.start_id'
+              'row_number() OVER (PARTITION BY r.group_id, r.priority <= ? ORDER BY r.start_id), r.priority, r.start_id',
+              [tailPriority]
             )
             .limit(want)
         : []

app/services/search_service.ts

@@ -304,9 +304,9 @@ class SearchService {
     // History by articles; the pass: the history ranges before those deferred behind them (priority
     // 2000 apart; e.g. groups added later that go first).
     const { rows: ranges } = await db.rawQuery(
-      `WITH h AS (SELECT status, priority, end_id - start_id + 1 AS n FROM ranges WHERE priority >= 0),
+      `WITH h AS (SELECT status, priority, end_id - start_id + 1 AS n FROM ranges WHERE priority >= 0 AND status <> 'ancient'),
             pass AS (SELECT (floor(coalesce(min(priority) FILTER (WHERE status <> 'done'), 0) / 2000) + 1) * 2000 AS below FROM h)
-       SELECT (SELECT count(*) FILTER (WHERE status = 'done') FROM ranges) AS done, (SELECT count(*) FROM ranges) AS total,
+       SELECT (SELECT count(*) FILTER (WHERE status = 'done') FROM ranges) AS done, (SELECT count(*) FROM ranges WHERE status <> 'ancient') AS total,
               coalesce(sum(n) FILTER (WHERE status = 'done'), 0) AS history_done, coalesce(sum(n), 0) AS history,
               coalesce(sum(n) FILTER (WHERE status = 'done' AND priority < (SELECT below FROM pass)), 0) AS pass_done,
               coalesce(sum(n) FILTER (WHERE priority < (SELECT below FROM pass)), 0) AS pass,

app/services/status.ts

@@ -0,0 +1,343 @@
+import db from '@adonisjs/lucid/services/db'
+import { readdir, stat, statfs } from 'node:fs/promises'
+import { join } from 'node:path'
+import coordinatorConfig from '#config/coordinator'
+import GroupLeases from '#services/group_leases'
+import type { Lane } from '#lib/edge/index'
+
+export interface RangeBand {
+  band: string
+  planned: number
+  plannedArticles: number
+  leased: number
+  done: number
+  doneArticles: number
+  failed: number
+}
+
+export interface HeldGroup {
+  node: string
+  group: string
+  lane: Lane
+  range: { start: number; end: number; next: number } | null
+  heartbeatSeconds: number
+  snapshotNext: number | null
+  snapshotSeconds: number | null
+  snapshotMb: number | null
+  rangesDone: number
+  headers: number
+}
+
+export interface SnapshotFile {
+  group: string
+  lane: Lane
+  mb: number
+  hours: number
+}
+
+export interface Status {
+  at: string
+  catalog: {
+    tablespace: string
+    gb: number
+    capGb: number
+    full: boolean
+    freeGb: number
+  }
+  bands: RangeBand[]
+  lastHour: {
+    indexedArticles: number
+    doneRanges: number
+    storedArticles: number
+    storedRanges: number
+  }
+  corpus: { pendingRanges: number; pendingArticles: number }
+  spool: { ranges: number; chunks: number }
+  tail: { groups: number; ranges: number; articles: number }
+  /** The 'ancient' ranges: articles below the low watermark the groups were planned at (expired, never fetched). */
+  ancient: { groups: number; articles: number }
+  held: HeldGroup[]
+  snapshots: { count: number; gb: number; oldestHours: number | null; largest: SnapshotFile[] }
+  releases: { estimated: number; provisional: number }
+  /** `ready` and `waiting` are the planner's estimates (the last analyze), the others counts. */
+  probes: { ready: number; waiting: number; pendingWaiting: number; pendingNamed: number }
+  judge: { dirty: number }
+  nodes: { name: string; lastSeenSeconds: number | null }[]
+}
+
+const seconds = (since: Date | string | null) =>
+  since ? Math.round((Date.now() - new Date(since).getTime()) / 1000) : null
+
+/**
+ * What the indexer is doing, in one object (commands/coordinator_status.ts prints it): the catalog's room on the
+ * NVMe, the ranges by band (the tail, the capture windows, the history), the last hour's rates, the held groups
+ * with their snapshots, the phase-2 backlogs. Every number comes from the database, the snapshot directory or the
+ * filesystem; nothing is kept between calls.
+ */
+class StatusService {
+  async collect(): Promise<Status> {
+    const tail = coordinatorConfig.tail.priority
+    const cap = coordinatorConfig.catalogMaxNvmeGb
+    // The catalog's tablespace only (CatalogSpace measures it every minute anyway): pg_database_size walks the
+    // chunk partitions and the probe cache on the HDD, minutes on homelab.
+    const { rows: space } = await db.rawQuery(
+      `SELECT coalesce(nullif(current_setting('default_tablespace'), ''), 'pg_default') AS tablespace,
+              pg_tablespace_size(coalesce(nullif(current_setting('default_tablespace'), ''), 'pg_default')) AS bytes`
+    )
+    const tablespaceGb = Number(space[0].bytes) / 1e9
+    const fs = await statfs('/').catch(() => null)
+    const freeGb = fs ? (fs.bavail * fs.bsize) / 1e9 : 0
+
+    // The ranges by band: the tail, the capture windows (priorities 0–2; the sampling ranges in other years are
+    // capture too, at their history priorities), the history by its thousand.
+    const { rows: bands } = await db.rawQuery(
+      `SELECT CASE WHEN priority <= ? THEN 'tail'
+                   WHEN priority < 0 THEN 'raised'
+                   WHEN capture AND priority <= 2 THEN 'window ' || priority
+                   WHEN priority < 1000 THEN 'history 0-999'
+                   ELSE 'history ' || (priority / 1000) * 1000 || '+' END AS band,
+              count(*) FILTER (WHERE status = 'planned') AS planned,
+              coalesce(sum(end_id - next_id + 1) FILTER (WHERE status = 'planned'), 0) AS planned_articles,
+              count(*) FILTER (WHERE status = 'leased') AS leased,
+              count(*) FILTER (WHERE status = 'done') AS done,
+              coalesce(sum(end_id - start_id + 1) FILTER (WHERE status = 'done'), 0) AS done_articles,
+              count(*) FILTER (WHERE status = 'failed') AS failed
+       FROM ranges GROUP BY 1
+       ORDER BY min(CASE WHEN priority <= ? THEN 0 WHEN priority < 0 THEN 1 WHEN capture AND priority <= 2 THEN 2 + priority
+                        WHEN priority < 1000 THEN 10 ELSE 11 + priority / 1000 END)`,
+      [tail, tail]
+    )
+    const { rows: hour } = await db.rawQuery(
+      `SELECT coalesce(sum(end_id - start_id + 1) FILTER (WHERE done_at > now() - interval '1 hour'), 0) AS indexed,
+              count(*) FILTER (WHERE done_at > now() - interval '1 hour') AS done_ranges,
+              coalesce(sum(end_id - start_id + 1) FILTER (WHERE stored_at > now() - interval '1 hour'), 0) AS stored,
+              count(*) FILTER (WHERE stored_at > now() - interval '1 hour') AS stored_ranges,
+              count(*) FILTER (WHERE capture AND stored_at IS NULL AND status <> 'done') AS corpus_pending,
+              coalesce(sum(end_id - start_id + 1) FILTER (WHERE capture AND stored_at IS NULL AND status <> 'done'), 0) AS corpus_pending_articles,
+              count(*) FILTER (WHERE status IN ('planned', 'leased') AND spooled_chunks > 0) AS spooled_ranges,
+              coalesce(sum(spooled_chunks) FILTER (WHERE status IN ('planned', 'leased')), 0) AS spooled_chunks,
+              count(DISTINCT group_id) FILTER (WHERE status = 'planned' AND priority <= ?) AS tail_groups,
+              count(*) FILTER (WHERE status = 'planned' AND priority <= ?) AS tail_ranges,
+              coalesce(sum(end_id - next_id + 1) FILTER (WHERE status = 'planned' AND priority <= ?), 0) AS tail_articles,
+              count(DISTINCT group_id) FILTER (WHERE status = 'ancient') AS ancient_groups,
+              coalesce(sum(end_id - start_id + 1) FILTER (WHERE status = 'ancient'), 0) AS ancient_articles
+       FROM ranges`,
+      [tail, tail, tail]
+    )
+    const h = hour[0]
+
+    const ttl = coordinatorConfig.leaseTtlSeconds
+    const { rows: held } = await db.rawQuery(
+      `SELECT n.name AS node, g.name AS "group", l.lane, l.heartbeat_at, l.snapshot_next, l.snapshot_at, l.ranges_done, l.headers,
+              r.start_id, r.end_id, r.next_id, l.group_id
+       FROM group_leases l JOIN groups g ON g.id = l.group_id JOIN nodes n ON n.id = l.node_id
+       LEFT JOIN ranges r ON r.id = l.range_id
+       WHERE l.node_id IS NOT NULL AND l.heartbeat_at >= now() - make_interval(secs => ?)
+       ORDER BY l.lane, g.name`,
+      [ttl]
+    )
+    const snapshotSizes = new Map<string, number>()
+    const files: SnapshotFile[] = []
+    const names = new Map<number, string>(
+      (await db.from('groups').select('id', 'name')).map((g) => [Number(g.id), g.name as string])
+    )
+    for (const file of await readdir(GroupLeases.dir).catch(() => [] as string[])) {
+      const m = /^(\d+)(-tail)?\.gz$/.exec(file)
+      if (!m) continue
+      const s = await stat(join(GroupLeases.dir, file)).catch(() => null)
+      if (!s) continue
+      const lane: Lane = m[2] ? 'tail' : 'history'
+      snapshotSizes.set(`${m[1]}:${lane}`, s.size)
+      files.push({
+        group: names.get(Number(m[1])) ?? m[1],
+        lane,
+        mb: s.size / 1e6,
+        hours: (Date.now() - s.mtimeMs) / 3600_000,
+      })
+    }
+    files.sort((a, b) => b.mb - a.mb)
+
+    // The probe backlog from the planner's estimates of the partial indexes (counting 11 M rows took 10 s on
+    // homelab); the rest are small partial indexes.
+    const { rows: releases } = await db.rawQuery(
+      `SELECT (SELECT reltuples::bigint FROM pg_class WHERE oid = 'releases'::regclass) AS estimated,
+              (SELECT count(*) FROM releases WHERE settled_at IS NULL) AS provisional,
+              (SELECT greatest(0, reltuples)::bigint FROM pg_class WHERE oid = 'probe_candidates_ready_idx'::regclass) AS ready,
+              (SELECT greatest(0, reltuples)::bigint FROM pg_class WHERE oid = 'probe_candidates_waiting_idx'::regclass) AS waiting,
+              (SELECT count(*) FROM pending_releases WHERE status = 0) AS pending_waiting,
+              (SELECT count(*) FROM pending_releases WHERE status = 4) AS pending_named,
+              (SELECT count(*) FROM judge_dirty) AS dirty`
+    )
+    const r = releases[0]
+    const nodes = await db.from('nodes').select('name', 'last_seen_at').orderBy('name')
+
+    return {
+      at: new Date().toISOString(),
+      catalog: {
+        tablespace: space[0].tablespace,
+        gb: tablespaceGb,
+        capGb: cap,
+        full: cap > 0 && tablespaceGb > cap,
+        freeGb,
+      },
+      bands: bands.map((b: any) => ({
+        band: b.band,
+        planned: Number(b.planned),
+        plannedArticles: Number(b.planned_articles),
+        leased: Number(b.leased),
+        done: Number(b.done),
+        doneArticles: Number(b.done_articles),
+        failed: Number(b.failed),
+      })),
+      lastHour: {
+        indexedArticles: Number(h.indexed),
+        doneRanges: Number(h.done_ranges),
+        storedArticles: Number(h.stored),
+        storedRanges: Number(h.stored_ranges),
+      },
+      corpus: {
+        pendingRanges: Number(h.corpus_pending),
+        pendingArticles: Number(h.corpus_pending_articles),
+      },
+      spool: { ranges: Number(h.spooled_ranges), chunks: Number(h.spooled_chunks) },
+      ancient: { groups: Number(h.ancient_groups), articles: Number(h.ancient_articles) },
+      tail: {
+        groups: Number(h.tail_groups),
+        ranges: Number(h.tail_ranges),
+        articles: Number(h.tail_articles),
+      },
+      held: held.map((l: any) => {
+        const size = snapshotSizes.get(`${l.group_id}:${l.lane}`)
+        return {
+          node: l.node,
+          group: l.group,
+          lane: l.lane,
+          range:
+            l.start_id !== null
+              ? { start: Number(l.start_id), end: Number(l.end_id), next: Number(l.next_id) }
+              : null,
+          heartbeatSeconds: seconds(l.heartbeat_at) ?? 0,
+          snapshotNext: l.snapshot_next === null ? null : Number(l.snapshot_next),
+          snapshotSeconds: seconds(l.snapshot_at),
+          snapshotMb: size === undefined ? null : size / 1e6,
+          rangesDone: Number(l.ranges_done),
+          headers: Number(l.headers),
+        }
+      }),
+      snapshots: {
+        count: files.length,
+        gb: files.reduce((sum, f) => sum + f.mb, 0) / 1000,
+        oldestHours: files.length > 0 ? Math.max(...files.map((f) => f.hours)) : null,
+        largest: files.slice(0, 5),
+      },
+      releases: { estimated: Math.max(0, Number(r.estimated)), provisional: Number(r.provisional) },
+      probes: {
+        ready: Number(r.ready),
+        waiting: Number(r.waiting),
+        pendingWaiting: Number(r.pending_waiting),
+        pendingNamed: Number(r.pending_named),
+      },
+      judge: { dirty: Number(r.dirty) },
+      nodes: nodes.map((n) => ({ name: n.name, lastSeenSeconds: seconds(n.last_seen_at) })),
+    }
+  }
+
+  /** The status as a text panel. */
+  render(s: Status): string {
+    const n = (v: number) => Math.round(v).toLocaleString('en-US')
+    const b = (v: number) =>
+      v >= 1e9
+        ? `${(v / 1e9).toFixed(2)} B`
+        : v >= 1e6
+          ? `${(v / 1e6).toFixed(1)} M`
+          : v >= 1e3
+            ? `${(v / 1e3).toFixed(0)} k`
+            : String(v)
+    const age = (sec: number | null) =>
+      sec === null
+        ? '-'
+        : sec < 90
+          ? `${sec} s`
+          : sec < 5400
+            ? `${Math.round(sec / 60)} min`
+            : `${(sec / 3600).toFixed(1)} h`
+    const lines: string[] = []
+    lines.push(`Status at ${s.at}`)
+    lines.push('')
+    lines.push(
+      `Catalog: ${s.catalog.gb.toFixed(1)} GB in tablespace ${s.catalog.tablespace}${s.catalog.capGb ? ` of ${s.catalog.capGb} GB${s.catalog.full ? ' (FULL: history paused)' : ''}` : ' (no cap)'}; ${s.catalog.freeGb.toFixed(1)} GB free on /`
+    )
+    lines.push(
+      `Last hour: ${b(s.lastHour.indexedArticles)} articles indexed in ${n(s.lastHour.doneRanges)} ranges (${n(s.lastHour.indexedArticles / 3600)}/s); corpus fill ${b(s.lastHour.storedArticles)} in ${n(s.lastHour.storedRanges)} ranges, ${n(s.corpus.pendingRanges)} capture ranges (${b(s.corpus.pendingArticles)}) still to store`
+    )
+    lines.push(
+      `Spool: ${n(s.spool.chunks)} chunks in ${n(s.spool.ranges)} ranges. Tail: ${b(s.tail.articles)} planned in ${n(s.tail.ranges)} ranges of ${n(s.tail.groups)} groups. Ancient: ${b(s.ancient.articles)} articles below the low watermark of ${n(s.ancient.groups)} groups`
+    )
+    lines.push('')
+    const table = (header: string[], rows: string[][]) => {
+      const widths = header.map((h, i) => Math.max(h.length, ...rows.map((r) => r[i].length)))
+      const line = (r: string[]) =>
+        r.map((c, i) => (i === 0 ? c.padEnd(widths[i]) : c.padStart(widths[i]))).join('  ')
+      lines.push(line(header))
+      for (const r of rows) lines.push(line(r))
+    }
+    lines.push('Ranges by band:')
+    table(
+      ['band', 'planned', 'articles', 'leased', 'done', 'articles', 'failed'],
+      s.bands.map((x) => [
+        x.band,
+        n(x.planned),
+        b(x.plannedArticles),
+        n(x.leased),
+        n(x.done),
+        b(x.doneArticles),
+        n(x.failed),
+      ])
+    )
+    lines.push('')
+    lines.push(`Held groups (${s.held.length}):`)
+    if (s.held.length > 0)
+      table(
+        [
+          'group',
+          'lane',
+          'node',
+          'range',
+          'at',
+          'heartbeat',
+          'snapshot',
+          'size',
+          'done',
+          'headers',
+        ],
+        s.held.map((x) => [
+          x.group,
+          x.lane,
+          x.node,
+          x.range ? `${x.range.start}-${x.range.end}` : '-',
+          x.range ? String(x.range.next) : '-',
+          age(x.heartbeatSeconds),
+          x.snapshotNext === null ? '-' : `${x.snapshotNext} (${age(x.snapshotSeconds)})`,
+          x.snapshotMb === null ? '-' : `${x.snapshotMb.toFixed(0)} MB`,
+          n(x.rangesDone),
+          b(x.headers),
+        ])
+      )
+    lines.push('')
+    lines.push(
+      `Snapshots: ${n(s.snapshots.count)} files, ${s.snapshots.gb.toFixed(2)} GB${s.snapshots.oldestHours === null ? '' : `, oldest ${s.snapshots.oldestHours.toFixed(1)} h`}${s.snapshots.largest.length > 0 ? `; largest ${s.snapshots.largest.map((f) => `${f.group}${f.lane === 'tail' ? ' (tail)' : ''} ${f.mb.toFixed(0)} MB`).join(', ')}` : ''}`
+    )
+    lines.push(
+      `Releases: ~${n(s.releases.estimated)}, ${n(s.releases.provisional)} provisional. Judge: ${n(s.judge.dirty)} dirty`
+    )
+    lines.push(
+      `Probes: ~${n(s.probes.ready)} ready of ~${n(s.probes.waiting)} waiting; pending ladder ${n(s.probes.pendingWaiting)} waiting, ${n(s.probes.pendingNamed)} named`
+    )
+    lines.push(
+      `Nodes: ${s.nodes.map((x) => `${x.name} (${age(x.lastSeenSeconds)})`).join(', ') || 'none'}`
+    )
+    return lines.join('\n')
+  }
+}
+
+export default new StatusService()

commands/coordinator_status.ts

@@ -0,0 +1,22 @@
+import { BaseCommand, flags } from '@adonisjs/core/ace'
+import type { CommandOptions } from '@adonisjs/core/types/ace'
+import StatusService from '#services/status'
+
+/**
+ * What the indexer is doing (app/services/status.ts): the catalog's room, the ranges by band, the last hour's
+ * rates, the held groups and their snapshots, the phase-2 backlogs. Reads only.
+ */
+export default class CoordinatorStatus extends BaseCommand {
+  static commandName = 'coordinator:status'
+  static description =
+    'Prints what the indexer is doing: catalog room, ranges by band, rates, held groups, backlogs'
+  static options: CommandOptions = { startApp: true }
+
+  @flags.boolean({ description: 'JSON instead of the text panel' })
+  declare json: boolean
+
+  async run() {
+    const status = await StatusService.collect()
+    this.logger.log(this.json ? JSON.stringify(status, null, 2) : StatusService.render(status))
+  }
+}

config/coordinator.ts

@@ -41,6 +41,12 @@ export 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),
+  /**
+   * How long a range's stream waits for a chunk the prefetcher hasn't spooled yet (app/controllers/node_controller.ts).
+   * Long: a chunk that keeps coming in isn't given up (only a stalled one is), and some groups' chunks take 20+
+   * minutes (a.b.0day.stuffz, 17 MB in 22 min; LOG.md 2026-10-10 (3)).
+   */
+  streamChunkWaitMs: env.get('STREAM_CHUNK_WAIT_MINUTES', 60) * 60 * 1000,
   /** 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),

database/migrations/1770000006600_group_lease_lanes.ts

@@ -0,0 +1,27 @@
+import { BaseSchema } from '@adonisjs/lucid/schema'
+
+/**
+ * Lease lanes (REBUILD.md, "Voyager 3: window indexing", 1, amended 2026-10-27): a group has two cursors, each with
+ * its own holder, window and snapshot. The `history` lane works the ranges the lease bounds hand out (the windows,
+ * the history), the `tail` lane the tail planner's (priority at or below the tail's): a thread on the tail never
+ * waits behind a group's history, and a held history group's tail isn't stuck behind its days of ranges.
+ */
+export default class extends BaseSchema {
+  async up() {
+    this.schema.raw(
+      `ALTER TABLE group_leases ADD COLUMN lane varchar(8) NOT NULL DEFAULT 'history'`
+    )
+    this.schema.raw(`ALTER TABLE group_leases DROP CONSTRAINT group_leases_pkey`)
+    this.schema.raw(`ALTER TABLE group_leases ADD PRIMARY KEY (group_id, lane)`)
+    // Counts the leases handed out: odd and even ones start from the other half of the order's mix.
+    this.schema.raw(`CREATE SEQUENCE group_lease_turns`)
+  }
+
+  async down() {
+    this.schema.raw(`DROP SEQUENCE IF EXISTS group_lease_turns`)
+    this.schema.raw(`DELETE FROM group_leases WHERE lane <> 'history'`)
+    this.schema.raw(`ALTER TABLE group_leases DROP CONSTRAINT group_leases_pkey`)
+    this.schema.raw(`ALTER TABLE group_leases DROP COLUMN lane`)
+    this.schema.raw(`ALTER TABLE group_leases ADD PRIMARY KEY (group_id)`)
+  }
+}

database/migrations/1770000006700_ancient_ranges.ts

@@ -0,0 +1,20 @@
+import { BaseSchema } from '@adonisjs/lucid/schema'
+
+/**
+ * One "ancient" range a group, articles 1 … (its first range − 1): what had expired on the server before the
+ * group was planned (the planner cuts from the low watermark). Status 'ancient': nothing to fetch, never leased,
+ * so the tiling is complete from article 1 and the span below the low watermark is on record (user, 2026-10-28).
+ * Planner.historyRows adds it for groups planned from now on.
+ */
+export default class extends BaseSchema {
+  async up() {
+    this.schema.raw(
+      `INSERT INTO ranges (group_id, start_id, end_id, next_id, priority, status)
+       SELECT group_id, 1, min(start_id) - 1, 1, 0, 'ancient' FROM ranges GROUP BY group_id HAVING min(start_id) > 1`
+    )
+  }
+
+  async down() {
+    this.schema.raw(`DELETE FROM ranges WHERE status = 'ancient'`)
+  }
+}

deploy/compose.yaml

@@ -178,6 +178,8 @@ services:
       # The window worker (REBUILD.md design 1 and 5): a group per thread, the window's snapshot to the coordinator
       # this often; the coordinator keeps them under SNAPSHOT_DIR (the spool's snapshots/ on bulk by default).
       EDGE_SNAPSHOT_MINUTES: ${EDGE_SNAPSHOT_MINUTES:-30}
+      # Of the threads, this many on the groups' tail lane (the live tail; the rest work the history lane).
+      EDGE_TAIL_THREADS: ${EDGE_TAIL_THREADS:-1}
     stop_grace_period: 60s
     depends_on:
       node-api:

edge/README.md

@@ -234,9 +234,20 @@ them:
 
 ```sh
 cargo build --release
-COORDINATOR_URL=http://coordinator:3334 NODE_TOKEN=… ./target/release/edge-node [--threads N] [--batch-articles 500000] [--snapshot-minutes 30] [--groups N] [--max-mbit M]
+COORDINATOR_URL=http://coordinator:3334 NODE_TOKEN=… ./target/release/edge-node [--threads N] [--tail-threads T] [--batch-articles 500000] [--snapshot-minutes 30] [--groups N] [--max-mbit M]
 ```
 
+- **Two lanes** (2026-10-27): `--tail-threads T` (EDGE_TAIL_THREADS; one of
+  two or more threads by default) puts that many threads on the groups'
+  *tail* lane (`?lane=tail` on the group routes): the tail planner's
+  ranges, through a window and snapshot of their own
+  (`<group>-tail.gz`). The other threads work the history lane: the
+  windows and the history the coordinator's bounds hand out. A tail thread
+  never wades through a group's history first, and a held history group's
+  tail isn't stuck behind its days of ranges; the coordinator orders each
+  lane's leases by alternating the group with the most planned articles
+  and the longest unheld one.
+
 - **One group per thread** (default one per core; REBUILD.md, "Voyager 3:
   window indexing", 1 and 5): a thread leases a group (`POST
   groups/lease`), takes its ranges in article order (`POST
@@ -273,5 +284,11 @@ COORDINATOR_URL=http://coordinator:3334 NODE_TOKEN=… ./target/release/edge-nod
   same (releases with every column, chunks, candidates, rule counts,
   fragments with their JSON).
 - **On homelab:** compose service `edge` (deploy/compose.yaml), the
-  `edge` image target of the Dockerfile, `--threads $EDGE_THREADS` (8);
+  `edge` image target of the Dockerfile, `--threads $EDGE_THREADS` (8),
-  it talks to `node-api`.
+  EDGE_TAIL_THREADS (1); it talks to `node-api`.
+- **End to end:** `tests/e2e/edge_node.spec.ts` runs this binary
+  (`edge/target/release/edge-node`, or EDGE_NODE_BIN) against the test
+  server, database, spool and corpus: a chunk the corpus fill stores while
+  the stream waits, a failed stream and the restart that restores the
+  snapshot, a restart whose snapshot file is gone, a takeover. `node ace
+  test e2e` after `cargo build --release`; CI builds the binary first.

edge/src/bin/edge-node.rs

@@ -5,7 +5,11 @@
 //! --snapshot-minutes and when the group is given back, and the next holder restores it. A group with no range
 //! left (caught up: the tail) is given back and leased again later.
 //!
-//!   COORDINATOR_URL=… NODE_TOKEN=… edge-node [--threads N] [--batch-articles 500000] [--snapshot-minutes 30] [--groups N] [--max-mbit M]
+//!   COORDINATOR_URL=… NODE_TOKEN=… edge-node [--threads N] [--tail-threads T] [--batch-articles 500000] [--snapshot-minutes 30] [--groups N] [--max-mbit M]
+//!
+//! `--tail-threads T` (EDGE_TAIL_THREADS; 1 of 2 or more threads by default): that many of the threads lease the
+//! groups' *tail* lane (`?lane=tail`: the tail planner's ranges, with a window and snapshot of their own), the rest
+//! the history lane (the windows and the history the coordinator's bounds hand out).
 //!
 //! SIGTERM/SIGINT: each thread finishes the chunk, posts its batch, snapshots and gives its group back.
 //! `--ranges`: the old range leases (POST lease), one range per thread without a carried window (kept for a
@@ -167,10 +171,10 @@ impl Api {
         }
     }
 
-    // Group leases (REBUILD.md design 1 and 5).
+    // Group leases (REBUILD.md design 1 and 5), per lane (`history` or `tail`).
 
-    fn lease_group(&self) -> Result<Option<GroupLeaseJson>, ApiError> {
+    fn lease_group(&self, lane: &str) -> Result<Option<GroupLeaseJson>, ApiError> {
-        let mut response = self.request("POST", "groups/lease", None, false, "application/json")?;
+        let mut response = self.request("POST", &format!("groups/lease?lane={lane}"), None, false, "application/json")?;
         match response.status().as_u16() {
             204 => Ok(None),
             200 => {
@@ -181,8 +185,8 @@ impl Api {
         }
     }
 
-    fn next_range(&self, group: i64) -> Result<Option<RangeSpanJson>, ApiError> {
+    fn next_range(&self, group: i64, lane: &str) -> Result<Option<RangeSpanJson>, ApiError> {
-        let mut response = self.request("POST", &format!("groups/{group}/next"), None, false, "application/json")?;
+        let mut response = self.request("POST", &format!("groups/{group}/next?lane={lane}"), None, false, "application/json")?;
         match response.status().as_u16() {
             204 => Ok(None),
             200 => {
@@ -193,8 +197,8 @@ impl Api {
         }
     }
 
-    fn group_call(&self, group: i64, action: &str) -> Result<(), ApiError> {
+    fn group_call(&self, group: i64, lane: &str, action: &str) -> Result<(), ApiError> {
-        let response = self.request("POST", &format!("groups/{group}/{action}"), None, false, "application/json")?;
+        let response = self.request("POST", &format!("groups/{group}/{action}?lane={lane}"), None, false, "application/json")?;
         match response.status().as_u16() {
             200..=299 => Ok(()),
             status => Err(ApiError::Other(format!("{action}: HTTP {status}"))),
@@ -202,11 +206,11 @@ impl Api {
     }
 
     /// The window's snapshot (bincode), gzipped, standing at `next`.
-    fn put_snapshot(&self, group: i64, next: i64, snapshot: &[u8]) -> Result<usize, ApiError> {
+    fn put_snapshot(&self, group: i64, lane: &str, next: i64, snapshot: &[u8]) -> Result<usize, ApiError> {
         let mut gz = GzEncoder::new(Vec::with_capacity(snapshot.len() / 3), flate2::Compression::new(1));
         gz.write_all(snapshot).unwrap();
         let body = gz.finish().unwrap();
-        let url = format!("{}/api/v1/node/groups/{group}/snapshot?next={next}", self.base);
+        let url = format!("{}/api/v1/node/groups/{group}/snapshot?next={next}&lane={lane}", self.base);
         let auth = format!("Bearer {}", self.token);
         for attempt in 1..=RETRIES {
             let result = self.agent.put(&url).header("Authorization", &auth).header("Content-Type", "application/octet-stream").send(&body[..]);
@@ -227,8 +231,8 @@ impl Api {
     }
 
     /// The group's snapshot: its bytes (inflated) and the article it stands at; None without one.
-    fn get_snapshot(&self, group: i64) -> Result<Option<(i64, Vec<u8>)>, ApiError> {
+    fn get_snapshot(&self, group: i64, lane: &str) -> Result<Option<(i64, Vec<u8>)>, ApiError> {
-        let response = self.request("GET", &format!("groups/{group}/snapshot"), None, false, "application/octet-stream")?;
+        let response = self.request("GET", &format!("groups/{group}/snapshot?lane={lane}"), None, false, "application/octet-stream")?;
         match response.status().as_u16() {
             404 => Ok(None),
             200 => {
@@ -544,9 +548,10 @@ enum RangeOutcome {
 }
 
 /// One range of a held group through the carried window: the chunks streamed, batches posted at the range's end at
-/// least, the range finished. The indexer keeps the window for the next range.
+/// least, the range finished. The indexer keeps the window for the next range. `group_name` is the label for the
+/// log (the tail lane's says so).
 #[allow(clippy::too_many_arguments)]
-fn run_group_range(api: &Arc<Api>, group_id: i64, group_name: &str, range: RangeSpanJson, indexer: &mut RangeIndexer, config: &Config, stopping: &AtomicBool, totals: &Arc<Mutex<Totals>>) -> RangeOutcome {
+fn run_group_range(api: &Arc<Api>, group_id: i64, lane: &str, group_name: &str, range: RangeSpanJson, indexer: &mut RangeIndexer, config: &Config, stopping: &AtomicBool, totals: &Arc<Mutex<Totals>>) -> RangeOutcome {
     let started = Instant::now();
     indexer.begin_range(range.next, range.end);
     let post_error: Arc<Mutex<Option<ApiError>>> = Arc::new(Mutex::new(None));
@@ -597,7 +602,7 @@ fn run_group_range(api: &Arc<Api>, group_id: i64, group_name: &str, range: Range
         }
         if since.elapsed().as_millis() as i64 - last_contact.load(Ordering::Relaxed) > HEARTBEAT.as_millis() as i64 {
             last_contact.store(since.elapsed().as_millis() as i64, Ordering::Relaxed);
-            let _ = api.group_call(group_id, "heartbeat");
+            let _ = api.group_call(group_id, lane, "heartbeat");
         }
         match frame {
             Frame::KeepAlive => continue,
@@ -706,9 +711,9 @@ fn run_group_range(api: &Arc<Api>, group_id: i64, group_name: &str, range: Range
 }
 
 /// The window's snapshot to the coordinator, standing at the indexer's next article.
-fn snapshot(api: &Api, group_id: i64, group_name: &str, indexer: &RangeIndexer) {
+fn snapshot(api: &Api, group_id: i64, lane: &str, group_name: &str, indexer: &RangeIndexer) {
     match indexer.snapshot() {
-        Ok(bytes) => match api.put_snapshot(group_id, indexer.next(), &bytes) {
+        Ok(bytes) => match api.put_snapshot(group_id, lane, indexer.next(), &bytes) {
             Ok(gz) => log("info", &format!("{group_name}: snapshot at {} ({:.1} MB, {:.1} MB gzipped).", indexer.next(), bytes.len() as f64 / 1e6, gz as f64 / 1e6)),
             Err(e) => log("warn", &format!("{group_name}: snapshot failed: {e}")),
         },
@@ -716,14 +721,16 @@ fn snapshot(api: &Api, group_id: i64, group_name: &str, indexer: &RangeIndexer)
     }
 }
 
-/// One held group: its window restored from the snapshot (or new), its ranges one after another until it is
+/// One held group (one lane of it): its window restored from the snapshot (or new), its ranges one after another
-/// caught up or the worker stops; then a snapshot, and the group given back.
+/// until it is caught up or the worker stops; then a snapshot, and the group given back.
-fn run_group(api: &Arc<Api>, lease: GroupLeaseJson, config: &Config, stopping: &AtomicBool) {
+fn run_group(api: &Arc<Api>, lease: GroupLeaseJson, lane: &str, config: &Config, stopping: &AtomicBool) {
     let group_id = lease.group.id;
-    let group_name = lease.group.name.clone();
+    let name = lease.group.name.clone();
+    // The log's label; the indexer gets the group's name itself (keys, xrefs).
+    let group_name = if lane == "tail" { format!("{name} (tail)") } else { name.clone() };
     let rules = lease.rules.rules();
     let mut indexer = match lease.snapshot {
-        Some(s) => match api.get_snapshot(group_id) {
+        Some(s) => match api.get_snapshot(group_id, lane) {
             Ok(Some((next, bytes))) => match RangeIndexer::restore(&bytes, rules.clone(), 5) {
                 Ok(ix) => {
                     log("info", &format!("{group_name}: window restored from the snapshot at {next} ({:.1} MB).", bytes.len() as f64 / 1e6));
@@ -731,20 +738,20 @@ fn run_group(api: &Arc<Api>, lease: GroupLeaseJson, config: &Config, stopping: &
                 }
                 Err(e) => {
                     log("warn", &format!("{group_name}: the snapshot at {} can't be restored ({e}); the window starts empty.", s.next));
-                    RangeIndexer::for_group(&group_name, rules.clone(), 5)
+                    RangeIndexer::for_group(&name, rules.clone(), 5)
                 }
             },
             Ok(None) => {
                 log("warn", &format!("{group_name}: the lease names a snapshot at {} but there is none; the window starts empty.", s.next));
-                RangeIndexer::for_group(&group_name, rules.clone(), 5)
+                RangeIndexer::for_group(&name, rules.clone(), 5)
             }
             Err(e) => {
                 log("warn", &format!("{group_name}: snapshot: {e}; the group goes back."));
-                let _ = api.group_call(group_id, "release");
+                let _ = api.group_call(group_id, lane, "release");
                 return;
             }
         },
-        None => RangeIndexer::for_group(&group_name, rules.clone(), 5),
+        None => RangeIndexer::for_group(&name, rules.clone(), 5),
     };
     let totals = Arc::new(Mutex::new(Totals::default()));
     let started = Instant::now();
@@ -754,7 +761,7 @@ fn run_group(api: &Arc<Api>, lease: GroupLeaseJson, config: &Config, stopping: &
         if stopping.load(Ordering::Relaxed) {
             break;
         }
-        let range = match api.next_range(group_id) {
+        let range = match api.next_range(group_id, lane) {
             Ok(Some(range)) => range,
             Ok(None) => {
                 log("info", &format!("{group_name}: caught up after {ranges} ranges; the group goes back."));
@@ -768,19 +775,19 @@ fn run_group(api: &Arc<Api>, lease: GroupLeaseJson, config: &Config, stopping: &
         if range.next != indexer.next() && ranges > 0 {
             log("warn", &format!("{group_name}: the window stands at {} but the next range resumes at {}: the window carries on from there.", indexer.next(), range.next));
         }
-        match run_group_range(api, group_id, &group_name, range, &mut indexer, config, stopping, &totals) {
+        match run_group_range(api, group_id, lane, &group_name, range, &mut indexer, config, stopping, &totals) {
             RangeOutcome::Done => ranges += 1,
             // The same range again from the reported chunk: the stream waits for its new copy.
             RangeOutcome::Again => {}
             RangeOutcome::Stop => break,
         }
         if snapshotted.elapsed() >= config.snapshot_every {
-            snapshot(api, group_id, &group_name, &indexer);
+            snapshot(api, group_id, lane, &group_name, &indexer);
             snapshotted = Instant::now();
         }
     }
-    snapshot(api, group_id, &group_name, &indexer);
+    snapshot(api, group_id, lane, &group_name, &indexer);
-    let _ = api.group_call(group_id, "release");
+    let _ = api.group_call(group_id, lane, "release");
     let t = totals.lock().unwrap();
     log(
         "info",
@@ -807,6 +814,12 @@ fn main() {
         .filter(|&n| n > 0)
         .unwrap_or_else(|| std::thread::available_parallelism().map(|n| n.get()).unwrap_or(1));
     let ranges: Option<usize> = flag("--ranges").map(|v| v.parse().unwrap());
+    // Of the threads, this many on the groups' tail lane: one by default when there are two or more.
+    let tail_threads: usize = flag("--tail-threads")
+        .or_else(|| std::env::var("EDGE_TAIL_THREADS").ok())
+        .map(|v| v.parse().unwrap())
+        .unwrap_or(if threads >= 2 { 1 } else { 0 });
+    assert!(tail_threads <= threads, "--tail-threads {tail_threads} is more than the {threads} threads");
     let max_mbit: Option<f64> = flag("--max-mbit").or_else(|| std::env::var("EDGE_MAX_MBIT").ok()).map(|v| v.parse().unwrap()).filter(|&m: &f64| m > 0.0);
     let config = Arc::new(Config {
         batch_articles: flag("--batch-articles").map(|v| v.parse().unwrap()).unwrap_or(500_000),
@@ -819,12 +832,13 @@ fn main() {
     signal_hook::flag::register(signal_hook::consts::SIGTERM, stopping.clone()).unwrap();
     signal_hook::flag::register(signal_hook::consts::SIGINT, stopping.clone()).unwrap();
     let api = Arc::new(Api::new(&url, &token));
-    log("info", &format!("Starting {threads} worker threads{}.", max_mbit.map(|m| format!(" (download cap {m} Mbit/s)")).unwrap_or_default()));
+    log("info", &format!("Starting {threads} worker threads{}{}.", if range_leases_wanted(&ranges) { String::new() } else { format!(", {tail_threads} on the tail") }, max_mbit.map(|m| format!(" (download cap {m} Mbit/s)")).unwrap_or_default()));
     let groups: Option<usize> = flag("--groups").map(|v| v.parse().unwrap());
     let range_leases = ranges.is_some();
     let workers: Vec<_> = (0..threads)
-        .map(|_| {
+        .map(|i| {
             let (api, config, stopping) = (api.clone(), config.clone(), stopping.clone());
+            let lane = if i < tail_threads { "tail" } else { "history" };
             std::thread::spawn(move || {
                 let mut done = 0;
                 if range_leases {
@@ -844,9 +858,9 @@ fn main() {
                     return;
                 }
                 while !stopping.load(Ordering::Relaxed) && groups.is_none_or(|n| done < n) {
-                    match api.lease_group() {
+                    match api.lease_group(lane) {
                         Ok(Some(lease)) => {
-                            run_group(&api, lease, &config, &stopping);
+                            run_group(&api, lease, lane, &config, &stopping);
                             done += 1;
                         }
                         Ok(None) if groups.is_some() => break,
@@ -865,6 +879,10 @@ fn main() {
     }
 }
 
+fn range_leases_wanted(ranges: &Option<usize>) -> bool {
+    ranges.is_some()
+}
+
 fn idle(stopping: &AtomicBool) {
     let until = Instant::now() + IDLE;
     while Instant::now() < until && !stopping.load(Ordering::Relaxed) {

lib/edge/index.ts

@@ -1,7 +1,7 @@
 export { RangeIndexer, postedSpan, chunkOf, candidatesOf, releaseKeyOf, uploadsOf, UPLOAD_GAP_MS, type EdgeBatch, type ReleaseChunk, type DroppedChunk, type ProbeCandidate, type CandidateKind, type RuleCount, type Fragment, type RangeIndexerOptions } from './range_indexer.js'
 export { encodeFrame, decodeFrames, isKeepalive, KEEPALIVE, type Frame } from './stream.js'
 export { toWire, fromWire, type WireBatch } from './wire.js'
-export type { Lease, GroupLease, NextRange } from './lease.js'
+export type { Lease, GroupLease, Lane, NextRange } from './lease.js'
 export type { EdgeSegment, FileMeta } from './scan.js'
 export { throttle } from './throttle.js'
 export { filterOverviews, undelta, type ChunkDrops } from './corpus_filter.js'

lib/edge/lease.ts

@@ -7,11 +7,20 @@ export interface Lease {
 }
 
 /**
- * A group lease (REBUILD.md, "Voyager 3: window indexing", 1): the group a worker thread holds, the rules, and the
+ * A group's two cursors (REBUILD.md, "Voyager 3: window indexing", 1): `history` works the ranges the lease bounds
- * snapshot of the group's window when there is one (its next article; the body comes from GET groups/:id/snapshot).
+ * hand out (the windows, the history), `tail` the tail planner's ranges. Each lane has its own holder, window and
+ * snapshot; the group routes take `?lane=` (history by default).
+ */
+export type Lane = 'history' | 'tail'
+
+/**
+ * A group lease (REBUILD.md, "Voyager 3: window indexing", 1): the group a worker thread holds, the lane, the rules,
+ * and the snapshot of the lane's window when there is one (its next article; the body comes from GET
+ * groups/:id/snapshot).
  */
 export interface GroupLease {
   group: { id: number; name: string; priority: number }
+  lane: Lane
   rules: Rules
   snapshot: { next: number } | null
 }

start/env.ts

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

tests/e2e/edge_node.spec.ts

@@ -0,0 +1,238 @@
+import { test } from '@japa/runner'
+import db from '@adonisjs/lucid/services/db'
+import testUtils from '@adonisjs/core/services/test_utils'
+import { mkdtemp, rm } from 'node:fs/promises'
+import { tmpdir } from 'node:os'
+import { join } from 'node:path'
+import Node from '#models/node'
+import Spool from '#services/spool'
+import Corpus from '#services/corpus'
+import GroupLeases from '#services/group_leases'
+import coordinatorConfig from '#config/coordinator'
+import { GROUP, overviewChunk, runEdge, startEdge, waitFor } from '#tests/e2e/helpers/edge'
+
+/**
+ * The Rust edge worker against the real node API, database, spool and corpus (REBUILD.md, "Voyager 3: window
+ * indexing", 1 and 5): what broke in production on 2026-10-27 (LOG.md (118)) — a stream waiting on a chunk the
+ * corpus fill stored, a failed stream, a restart with and without the snapshot, a takeover.
+ */
+test.group('Edge worker end to end', (group) => {
+  const saved = {
+    spool: coordinatorConfig.spool.dir,
+    corpus: coordinatorConfig.corpus.dir,
+    snapshots: coordinatorConfig.snapshotDir,
+    chunk: coordinatorConfig.chunkArticles,
+    wait: coordinatorConfig.streamChunkWaitMs,
+  }
+  let dir = ''
+  group.each.setup(async () => {
+    dir = await mkdtemp(join(tmpdir(), 'e2e-'))
+    coordinatorConfig.spool.dir = join(dir, 'spool')
+    coordinatorConfig.corpus.dir = join(dir, 'corpus')
+    coordinatorConfig.snapshotDir = join(dir, 'snapshots')
+    coordinatorConfig.chunkArticles = 100
+    return testUtils.db().truncate()
+  })
+  group.each.teardown(async () => {
+    Object.assign(coordinatorConfig.spool, { dir: saved.spool })
+    coordinatorConfig.corpus.dir = saved.corpus
+    coordinatorConfig.snapshotDir = saved.snapshots
+    coordinatorConfig.chunkArticles = saved.chunk
+    coordinatorConfig.streamChunkWaitMs = saved.wait
+    await rm(dir, { recursive: true, force: true })
+  })
+
+  /** A group with one range of three chunks and a node with a token. */
+  async function setup(opts: { capture?: boolean } = {}) {
+    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,
+        priority: 0,
+        capture: opts.capture ?? false,
+      })
+      .returning('id')
+    const node = await Node.create({ name: 'e2e-node' })
+    const token = (await Node.accessTokens.create(node)).value!.release()
+    return { groupId: Number(g.id), rangeId: Number(r.id), nodeId: Number(node.id), token }
+  }
+  const range = async (id: number) => (await db.from('ranges').where('id', id).firstOrFail()) as any
+  const lease = async (groupId: number) =>
+    (await db
+      .from('group_leases')
+      .where('group_id', groupId)
+      .where('lane', 'history')
+      .first()) as any
+  /** The release whose first article is `at` (a release's name isn't the file's: the identity strips it). */
+  const releaseAt = async (at: number) =>
+    (await db.from('releases').where('first_article', at).first()) as any
+  const names = async () =>
+    JSON.stringify(
+      await db
+        .from('releases')
+        .select('name', 'first_article', 'last_article', 'segments', 'complete')
+        .orderBy('first_article')
+    )
+  const has = (lines: string[], text: string) => lines.some((l) => l.includes(text))
+
+  test('a range streamed from the spool and the corpus, with a chunk the corpus fill stores while the stream waits', async ({
+    assert,
+  }) => {
+    const { groupId, rangeId, token } = await setup({ capture: true })
+    await Spool.write(rangeId, { from: 1, to: 100 }, overviewChunk(1, 100))
+    await Spool.write(rangeId, { from: 201, to: 300 }, overviewChunk(201, 300))
+    const run = await startEdge(token)
+    // The worker holds the group and streams: chunk 1 goes out, chunk 2 is waited for — the corpus fill stores it.
+    await waitFor(
+      'the group to be held',
+      async () =>
+        (await lease(groupId))?.range_id !== null && (await lease(groupId))?.range_id !== undefined
+    )
+    await new Promise((r) => setTimeout(r, 700))
+    await Corpus.write(
+      { group: GROUP, start: 1, end: 300 },
+      { from: 101, to: 200 },
+      overviewChunk(101, 200),
+      3
+    )
+    const { code, lines } = { ...(await run.exit), lines: run.lines }
+    assert.equal(code, 0, lines.join('\n'))
+    assert.isTrue(has(lines, 'caught up after 1 ranges'), lines.join('\n'))
+    assert.isTrue(has(lines, 'snapshot at 301'), lines.join('\n'))
+    assert.include(await range(rangeId), { status: 'done' })
+    assert.equal(Number((await range(rangeId)).headers), 300)
+    const l = await lease(groupId)
+    assert.isNull(l.node_id)
+    assert.equal(Number(l.snapshot_next), 301)
+    assert.equal(Number(l.ranges_done), 1)
+    // The window closed on the early files (the frontier is a day past them); every closed release is settled.
+    const releases = await db
+      .from('releases')
+      .select('name', 'settled_at', 'first_article', 'last_article', 'segments')
+    assert.isAtLeast(releases.length, 10)
+    assert.isTrue(releases.every((r) => r.settled_at !== null))
+    const spanning = releases.find((r) => Number(r.first_article) === 100)
+    assert.isDefined(spanning, await names())
+    assert.deepEqual(
+      [Number(spanning.first_article), Number(spanning.last_article), Number(spanning.segments)],
+      [100, 101, 2],
+      'the file over the chunk boundary came through the carried window'
+    )
+  })
+
+  test('a stream that fails: the batch so far is posted, the window snapshotted, the group given back; the next run restores it', async ({
+    assert,
+  }) => {
+    coordinatorConfig.streamChunkWaitMs = 1500
+    const { groupId, rangeId, token } = await setup()
+    await Spool.write(rangeId, { from: 1, to: 100 }, overviewChunk(1, 100))
+    const first = await runEdge(token)
+    assert.equal(first.code, 0, first.lines.join('\n'))
+    assert.isTrue(has(first.lines, 'the group goes back'), first.lines.join('\n'))
+    assert.isTrue(has(first.lines, 'snapshot at 101'), first.lines.join('\n'))
+    let r = await range(rangeId)
+    assert.include(r, { status: 'planned', node_id: null })
+    assert.equal(
+      Number(r.next_id),
+      101,
+      'the batch for chunk 1 was stored before the group went back'
+    )
+    let l = await lease(groupId)
+    assert.isNull(l.node_id)
+    assert.equal(Number(l.snapshot_next), 101)
+    const before = await db.from('releases').count('* as n').first()
+    assert.isAtLeast(Number(before!.n), 10)
+    assert.isNull(await releaseAt(100), 'the file over the boundary is still open in the window')
+
+    // The next run: the window restored, the rest of the range streamed, the open file completed.
+    coordinatorConfig.streamChunkWaitMs = saved.wait
+    await Spool.write(rangeId, { from: 101, to: 200 }, overviewChunk(101, 200))
+    await Spool.write(rangeId, { from: 201, to: 300 }, overviewChunk(201, 300))
+    const second = await runEdge(token)
+    assert.equal(second.code, 0, second.lines.join('\n'))
+    assert.isTrue(
+      has(second.lines, 'window restored from the snapshot at 101'),
+      second.lines.join('\n')
+    )
+    r = await range(rangeId)
+    assert.equal(r.status, 'done')
+    assert.equal(Number(r.headers), 300)
+    l = await lease(groupId)
+    assert.equal(Number(l.snapshot_next), 301)
+    const spanning = await releaseAt(100)
+    assert.isNotNull(spanning, await names())
+    assert.deepEqual(
+      [Number(spanning.first_article), Number(spanning.last_article), Number(spanning.segments)],
+      [100, 101, 2]
+    )
+  })
+
+  test('a restart whose snapshot file is gone: the window starts empty and the file over the seam ends incomplete', async ({
+    assert,
+  }) => {
+    coordinatorConfig.streamChunkWaitMs = 1500
+    const { groupId, rangeId, token } = await setup()
+    await Spool.write(rangeId, { from: 1, to: 100 }, overviewChunk(1, 100))
+    const first = await runEdge(token)
+    assert.isTrue(has(first.lines, 'snapshot at 101'), first.lines.join('\n'))
+    await rm(GroupLeases.snapshotPath(groupId), { force: true })
+
+    coordinatorConfig.streamChunkWaitMs = saved.wait
+    await Spool.write(rangeId, { from: 101, to: 200 }, overviewChunk(101, 200))
+    await Spool.write(rangeId, { from: 201, to: 300 }, overviewChunk(201, 300))
+    const second = await runEdge(token)
+    assert.equal(second.code, 0, second.lines.join('\n'))
+    assert.isTrue(
+      has(second.lines, 'but there is none; the window starts empty'),
+      second.lines.join('\n')
+    )
+    assert.equal((await range(rangeId)).status, 'done')
+    // Part 1 was in the lost window: one seam. Part 2 alone is a file short of a part.
+    assert.isNull(await releaseAt(100), await names())
+    const half = await releaseAt(101)
+    assert.isNotNull(half, await names())
+    assert.deepEqual([Number(half.segments), half.complete], [1, false], await names())
+  })
+
+  test("a takeover: the stale holder's next batch is refused, its snapshot too, and the group stays the new holder's", async ({
+    assert,
+  }) => {
+    const { groupId, rangeId, token, nodeId } = await setup()
+    await Spool.write(rangeId, { from: 1, to: 100 }, overviewChunk(1, 100))
+    const run = await startEdge(token)
+    // Chunk 1 posted; the stream now waits for chunk 2.
+    await waitFor('the first batch', async () => Number((await range(rangeId)).next_id) === 101)
+    // The holder looks dead (its group and range heartbeats stale): another node takes the group over (no snapshot
+    // yet: the range goes back to its resume point).
+    const stale = new Date(Date.now() - 24 * 3600_000)
+    await db.from('group_leases').where('group_id', groupId).update({ heartbeat_at: stale })
+    await db.from('ranges').where('id', rangeId).update({ heartbeat_at: stale })
+    const other = await Node.create({ name: 'e2e-other' })
+    const taken = await GroupLeases.lease(Number(other.id))
+    assert.equal(taken?.group.id, groupId)
+    assert.isNull(taken?.snapshot)
+    assert.include(await range(rangeId), { status: 'planned', node_id: null })
+    // The old holder gets chunk 2 all the same, posts, and is refused.
+    await Spool.write(rangeId, { from: 101, to: 200 }, overviewChunk(101, 200))
+    await Spool.write(rangeId, { from: 201, to: 300 }, overviewChunk(201, 300))
+    const { code } = await run.exit
+    const lines = run.lines
+    assert.equal(code, 0, lines.join('\n'))
+    assert.isTrue(has(lines, 'snapshot failed'), lines.join('\n'))
+    assert.isTrue(
+      has(lines, 'not leased to this node') || has(lines, 'no longer this node'),
+      lines.join('\n')
+    )
+    const l = await lease(groupId)
+    assert.equal(Number(l.node_id), Number(other.id))
+    assert.isNull(l.snapshot_next)
+    const r = await range(rangeId)
+    assert.equal(r.status, 'planned')
+    assert.equal(Number(r.next_id), 101, 'nothing past the first batch was stored')
+    assert.notEqual(Number(r.node_id), nodeId)
+  })
+})

tests/e2e/helpers/edge.ts

@@ -0,0 +1,149 @@
+import { spawn } from 'node:child_process'
+import { access } from 'node:fs/promises'
+import { resolve } from 'node:path'
+import { deflateSync } from 'node:zlib'
+import env from '#start/env'
+
+export const GROUP = 'alt.binaries.e2e'
+
+/** The edge worker binary: EDGE_NODE_BIN, else the release or debug build under edge/target. */
+export async function edgeBinary(): Promise<string> {
+  const candidates = [
+    process.env.EDGE_NODE_BIN,
+    'edge/target/release/edge-node',
+    'edge/target/debug/edge-node',
+  ]
+    .filter((c): c is string => !!c)
+    .map((c) => resolve(c))
+  for (const path of candidates) {
+    try {
+      await access(path)
+      return path
+    } catch {}
+  }
+  throw new Error(
+    `No edge-node binary (looked at ${candidates.join(', ')}): build it with "cargo build --release --manifest-path edge/Cargo.toml --bin edge-node", or set EDGE_NODE_BIN`
+  )
+}
+
+/** The test server's address for the worker (tests/bootstrap.ts starts it on HOST:PORT). */
+export function coordinatorUrl(): string {
+  const host = env.get('HOST', 'localhost')
+  return `http://${host === '0.0.0.0' ? '127.0.0.1' : host}:${env.get('PORT', 3333)}`
+}
+
+export interface EdgeRun {
+  exit: Promise<{ code: number | null; signal: string | null }>
+  lines: string[]
+  kill: () => void
+}
+
+/**
+ * The real edge worker against the test server: one thread on the history lane, one group, a batch per chunk
+ * (100 articles in these tests), a snapshot after every range.
+ */
+export async function startEdge(token: string, args: string[] = []): Promise<EdgeRun> {
+  const bin = await edgeBinary()
+  // The test server first (tests/bootstrap.ts starts it for this suite): a worker that can't connect retries for
+  // minutes before it says so.
+  const probe = await fetch(`${coordinatorUrl()}/api/v1/node/lease`, { method: 'POST' }).catch(
+    (e) => e as Error
+  )
+  if (probe instanceof Error || probe.status !== 401)
+    throw new Error(
+      `The test server doesn't answer at ${coordinatorUrl()}: ${probe instanceof Error ? probe.message : `HTTP ${probe.status}`}`
+    )
+  const child = spawn(
+    bin,
+    [
+      '--threads',
+      '1',
+      '--tail-threads',
+      '0',
+      '--groups',
+      '1',
+      '--batch-articles',
+      '100',
+      '--snapshot-minutes',
+      '0',
+      ...args,
+    ],
+    {
+      env: { ...process.env, COORDINATOR_URL: coordinatorUrl(), NODE_TOKEN: token },
+      stdio: ['ignore', 'pipe', 'pipe'],
+    }
+  )
+  const lines: string[] = []
+  let rest = ''
+  const collect = (chunk: Buffer) => {
+    rest += chunk.toString('utf8')
+    const parts = rest.split('\n')
+    rest = parts.pop() ?? ''
+    lines.push(...parts)
+  }
+  child.stdout.on('data', collect)
+  child.stderr.on('data', collect)
+  const exit = new Promise<{ code: number | null; signal: string | null }>((done) => {
+    child.on('exit', (code, signal) => {
+      if (rest) lines.push(rest)
+      done({ code, signal })
+    })
+  })
+  return { exit, lines, kill: () => child.kill('SIGKILL') }
+}
+
+/** Runs the worker to its end (at most `timeoutMs`), its log lines and exit code. */
+export async function runEdge(token: string, args: string[] = [], timeoutMs = 60_000) {
+  const run = await startEdge(token, args)
+  const timer = setTimeout(() => run.kill(), timeoutMs)
+  const result = await run.exit
+  clearTimeout(timer)
+  return { ...result, lines: run.lines }
+}
+
+/** Polls `condition` every 100 ms until it holds or `timeoutMs` passed (then throws). */
+export async function waitFor(what: string, condition: () => Promise<boolean>, timeoutMs = 30_000) {
+  const until = Date.now() + timeoutMs
+  while (Date.now() < until) {
+    if (await condition()) return
+    await new Promise((r) => setTimeout(r, 100))
+  }
+  throw new Error(`Timed out waiting for ${what}`)
+}
+
+const BASE_MS = Date.UTC(2026, 9, 5, 10)
+
+/** A file name without digits (a number sequence would make an album of a poster's files): letters from `k`. */
+export function nameOf(k: number): string {
+  let s = ''
+  let n = k
+  do {
+    s = String.fromCharCode(97 + (n % 26)) + s
+    n = Math.floor(n / 26)
+  } while (n > 0)
+  return `Set.${s}.mkv`
+}
+
+/**
+ * The overview lines of articles `from`..`to`, one an hour from 2026-10-05: article 1 is a one-part file, and from 2
+ * on each even article is part 1 of a two-part file whose part 2 is the next (odd) article — so a file spans every
+ * chunk boundary at an even `to` (100 and 101, 200 and 201: the window has to carry it). Dates an hour apart make
+ * the posting frontier move a day every 24 articles, so the window closes on files as it goes.
+ */
+export function overviewLines(from: number, to: number): string[] {
+  const lines: string[] = []
+  for (let n = from; n <= to; n++) {
+    const date = new Date(BASE_MS + n * 3600_000).toUTCString().replace(' GMT', ' +0000')
+    const [name, part] =
+      n === 1 ? [nameOf(1), '(1/1)'] : n % 2 === 0 ? [nameOf(n), '(1/2)'] : [nameOf(n - 1), '(2/2)']
+    lines.push(
+      `${n}\t"${name}" yEnc ${part}\tposter <p@x>\t${date}\t<a${n}@e2e>\t\t700000\t5000\tXref: news.example ${GROUP}:${n}`
+    )
+  }
+  return lines
+}
+
+/** One XOVER chunk as the news server sends it (deflated). */
+export function overviewChunk(from: number, to: number): Buffer {
+  return deflateSync(Buffer.from(overviewLines(from, to).join('\r\n') + '\r\n'))
+}

tests/functional/group_leases.spec.ts

@@ -42,7 +42,7 @@ test.group('Group leases', (g) => {
     await rm(dir, { recursive: true, force: true })
   })
 
-  test('a lease takes the group with the most urgent planned range; its ranges come in article order; caught up it gives nothing', async ({
+  test('a lease takes the group with the most urgent planned range of its lane; its ranges come in article order; caught up it gives nothing', async ({
     assert,
   }) => {
     const a = await group('alt.binaries.gl.a', 1)
@@ -56,33 +56,73 @@ test.group('Group leases', (g) => {
     const n1 = await node('gl-1')
     const n2 = await node('gl-2')
 
+    // The history lane: both groups have priority 0 work; a has the most planned articles (300 to 100).
     const first = await GroupLeases.lease(n1)
-    assert.equal(first?.group.id, b, 'the tail first')
+    assert.equal(first?.group.id, a, 'the most planned articles first')
+    assert.equal(first?.lane, 'history')
     assert.isNull(first?.snapshot)
     // Held: the next node gets the other group; a third gets nothing.
     const second = await GroupLeases.lease(n2)
-    assert.equal(second?.group.id, a)
+    assert.equal(second?.group.id, b)
     assert.isNull(await GroupLeases.lease(await node('gl-3')))
 
-    // Ranges in article order, whatever their priority.
+    // Ranges in article order, whatever their priority; the tail range isn't the history lane's.
-    assert.deepEqual((await GroupLeases.next(b, n1))?.range, { id: b1, start: 1, end: 100, next: 1 })
+    assert.deepEqual((await GroupLeases.next(b, n2))?.range, { id: b2, start: 101, end: 200, next: 101 })
-    assert.include(await rangeRow(b1), { status: 'leased', node_id: n1 })
+    assert.include(await rangeRow(b2), { status: 'leased', node_id: n2 })
     // Asked again before finishing: the same range.
-    assert.equal((await GroupLeases.next(b, n1))?.range.id, b1)
+    assert.equal((await GroupLeases.next(b, n2))?.range.id, b2)
-    await db.from('ranges').where('id', b1).update({ status: 'done', next_id: 101 })
-    assert.equal((await GroupLeases.next(b, n1))?.range.id, b2)
     // A chunk of b2 that doesn't inflate, reported after the batch before it: the range is asked for again and
     // comes back at its resume point (the chunk fetched plain next time).
     await db.from('ranges').where('id', b2).update({ next_id: 151 })
-    await Leases.corrupt(b2, n1, 151)
+    await Leases.corrupt(b2, n2, 151)
     assert.include(await rangeRow(b2), { status: 'planned', node_id: null })
-    assert.deepEqual((await GroupLeases.next(b, n1))?.range, { id: b2, start: 101, end: 200, next: 151 })
+    assert.deepEqual((await GroupLeases.next(b, n2))?.range, { id: b2, start: 101, end: 200, next: 151 })
     assert.deepEqual((await rangeRow(b2)).plain_chunks.map(Number), [151])
     await db.from('ranges').where('id', b2).update({ status: 'done', next_id: 201 })
-    assert.isNull(await GroupLeases.next(b, n1), 'caught up')
+    assert.isNull(await GroupLeases.next(b, n2), 'caught up: the tail range is the other lane\'s')
+    assert.include(await rangeRow(b1), { status: 'planned' })
+
+    // The tail lane: group b's tail, held by a third node while n2 holds b's history.
+    const n3 = await node('gl-3b')
+    const tail = await GroupLeases.lease(n3, 'tail')
+    assert.equal(tail?.group.id, b)
+    assert.equal(tail?.lane, 'tail')
+    assert.isNull(await GroupLeases.lease(await node('gl-3c'), 'tail'), 'no other group has tail work')
+    assert.deepEqual((await GroupLeases.next(b, n3, 'tail'))?.range, { id: b1, start: 1, end: 100, next: 1 })
+    // The history lane is n2's.
+    await assert.rejects(() => GroupLeases.next(b, n3), LeaseLost)
+    await db.from('ranges').where('id', b1).update({ status: 'done', next_id: 101 })
+    assert.isNull(await GroupLeases.next(b, n3, 'tail'), 'the tail lane is caught up')
+    // Each lane its own snapshot.
+    await GroupLeases.putSnapshot(b, n3, 101, Readable.from([Buffer.from('tail at 101')]), 'tail')
+    await GroupLeases.putSnapshot(b, n2, 201, Readable.from([Buffer.from('history at 201')]))
+    assert.equal((await GroupLeases.getSnapshot(b, n3, 'tail'))?.next, 101)
+    assert.equal((await GroupLeases.getSnapshot(b, n2))?.next, 201)
+    await GroupLeases.release(b, n3, 'tail')
+    assert.include(await GroupLeases.held(b, n2), { node_id: n2 })
+    await assert.rejects(() => GroupLeases.held(b, n3, 'tail'), LeaseLost)
 
     assert.deepEqual([a1, a2, a3].length, 3)
-    await assert.rejects(() => GroupLeases.next(a, n1), LeaseLost)
+    await assert.rejects(() => GroupLeases.next(a, n2), LeaseLost)
+  })
+
+  test('the order mixes the group with the most planned articles and the longest unheld one', async ({ assert }) => {
+    // Four groups with priority 0 work: c was held before (5000 articles left), a, b and d never (100, 1000, 50).
+    const a = await group('alt.binaries.gl.o.a', 1)
+    const b = await group('alt.binaries.gl.o.b', 2)
+    const c = await group('alt.binaries.gl.o.c', 3)
+    const d = await group('alt.binaries.gl.o.d', 4)
+    await range(a, 1, 0)
+    for (let i = 0; i < 10; i++) await range(b, 1 + 100 * i, 0)
+    for (let i = 0; i < 50; i++) await range(c, 1 + 100 * i, 0)
+    await db.table('ranges').insert({ group_id: d, start_id: 1, end_id: 50, next_id: 1, priority: 0 })
+    await db.table('group_leases').insert({ group_id: c, lane: 'history', node_id: null, heartbeat_at: new Date(Date.now() - 3600_000) })
+    // By articles c, b, a, d; by age a, b, d (never held, by group priority), c. The leases alternate which
+    // ranking goes first: c, a, b, d (or a, c, b, d when the sequence stands on the other parity).
+    await db.rawQuery(`SELECT setval('group_lease_turns', 2)`)
+    const order = []
+    for (const name of ['o1', 'o2', 'o3', 'o4']) order.push((await GroupLeases.lease(await node(name)))?.group.id)
+    assert.deepEqual(order, [c, a, b, d])
   })
 
   test('a stale holder is taken over and the ranges after the snapshot are read again; a release keeps the resume point', async ({
@@ -165,14 +205,19 @@ test.group('Group leases', (g) => {
     const r2 = await range(a, 101, 0)
     const n1 = await node('gl-6')
     await GroupLeases.lease(n1)
+    // A range after the snapshot's article done long before the snapshot was taken (the old range leases did it):
+    // not the window's, it stays done.
+    const r3 = await range(a, 201, 0)
+    await db.from('ranges').where('id', r3).update({ status: 'done', next_id: 301, done_at: new Date(Date.now() - 48 * 3600_000) })
     // The worker snapshotted at the start, finished r1, and crashed before its next snapshot: r1 is read again.
     await GroupLeases.putSnapshot(a, n1, 1, Readable.from([Buffer.from('window at 1')]))
-    await db.from('ranges').where('id', r1).update({ status: 'done', next_id: 101 })
+    await db.from('ranges').where('id', r1).update({ status: 'done', next_id: 101, done_at: new Date() })
     await db.from('group_leases').where('group_id', a).update({ heartbeat_at: new Date(Date.now() - 24 * 3600_000) })
     await GroupLeases.lease(await node('gl-7'))
     assert.equal((await rangeRow(r1)).status, 'planned')
     assert.equal(Number((await rangeRow(r1)).next_id), 1)
     assert.equal((await rangeRow(r2)).status, 'planned')
     assert.equal(Number((await rangeRow(r2)).next_id), 101)
+    assert.equal((await rangeRow(r3)).status, 'done')
   })
 })

tests/functional/node_api.spec.ts

@@ -145,6 +145,23 @@ test.group('Node API', (group) => {
       ;(await client.post(`/api/v1/node/groups/${groupId}/release`).bearerToken(token)).assertStatus(200)
       ;(await client.post(`/api/v1/node/groups/${groupId}/heartbeat`).bearerToken(token)).assertStatus(409)
       assert.include(await db.from('ranges').where('id', r2.id).firstOrFail(), { status: 'planned', node_id: null })
+
+      // The tail lane (`?lane=tail`): the group's tail range, with a cursor and snapshot of its own.
+      const [t] = await db.table('ranges').insert({ group_id: g.id, start_id: 401, end_id: 500, next_id: 401, priority: -3 }).returning('id')
+      ;(await client.post('/api/v1/node/groups/lease?lane=sideways').bearerToken(token)).assertStatus(400)
+      const tail = await client.post('/api/v1/node/groups/lease?lane=tail').bearerToken(token)
+      tail.assertStatus(200)
+      assert.equal(tail.body().lane, 'tail')
+      assert.equal(Number(tail.body().group.id), groupId)
+      assert.deepEqual((await client.post(`/api/v1/node/groups/${groupId}/next?lane=tail`).bearerToken(token)).body().range, { id: Number(t.id), start: 401, end: 500, next: 401 })
+      ;(await client.post(`/api/v1/node/groups/${groupId}/heartbeat`).bearerToken(token)).assertStatus(409)
+      ;(await client.post(`/api/v1/node/groups/${groupId}/heartbeat?lane=tail`).bearerToken(token)).assertStatus(200)
+      ;(await putRaw(`/api/v1/node/groups/${groupId}/snapshot?next=401&lane=tail`, Buffer.from('tail at 401'))).assertStatus(200)
+      const gotTail = await client.get(`/api/v1/node/groups/${groupId}/snapshot?lane=tail`).bearerToken(token)
+      gotTail.assertStatus(200)
+      assert.equal(gotTail.header('x-snapshot-next'), '401')
+      ;(await client.post(`/api/v1/node/groups/${groupId}/release?lane=tail`).bearerToken(token)).assertStatus(200)
+      assert.include(await db.from('ranges').where('id', t.id).firstOrFail(), { status: 'planned', node_id: null })
     } finally {
       coordinatorConfig.corpus.dir = saved.corpus
       coordinatorConfig.snapshotDir = saved.snapshots

tests/functional/status.spec.ts

@@ -0,0 +1,168 @@
+import { test } from '@japa/runner'
+import db from '@adonisjs/lucid/services/db'
+import testUtils from '@adonisjs/core/services/test_utils'
+import { mkdtemp, rm, writeFile } from 'node:fs/promises'
+import { tmpdir } from 'node:os'
+import { join } from 'node:path'
+import coordinatorConfig from '#config/coordinator'
+import StatusService from '#services/status'
+
+test.group('Status', (group) => {
+  let dir = ''
+  let snapshotDir = ''
+  group.each.setup(async () => {
+    const rollback = await testUtils.db().withGlobalTransaction()
+    dir = await mkdtemp(join(tmpdir(), 'snapshots-'))
+    snapshotDir = coordinatorConfig.snapshotDir
+    coordinatorConfig.snapshotDir = dir
+    return rollback
+  })
+  group.each.teardown(async () => {
+    coordinatorConfig.snapshotDir = snapshotDir
+    await rm(dir, { recursive: true, force: true })
+  })
+
+  test('the ranges by band, the held groups with their snapshots, the backlogs', async ({
+    assert,
+  }) => {
+    const [g] = await db
+      .table('groups')
+      .insert({ name: 'alt.binaries.status', priority: 1 })
+      .returning('id')
+    const [n] = await db
+      .table('nodes')
+      .insert({ name: 'status-node', last_seen_at: new Date() })
+      .returning('id')
+    const ranges = [
+      { start_id: 1, end_id: 100, next_id: 1, priority: -3 },
+      { start_id: 101, end_id: 200, next_id: 101, priority: -1 },
+      {
+        start_id: 201,
+        end_id: 300,
+        next_id: 301,
+        priority: 0,
+        capture: true,
+        status: 'done',
+        done_at: new Date(),
+      },
+      {
+        start_id: 301,
+        end_id: 400,
+        next_id: 351,
+        priority: 0,
+        capture: true,
+        status: 'leased',
+        node_id: n.id,
+        spooled_chunks: 2,
+      },
+      {
+        start_id: 401,
+        end_id: 500,
+        next_id: 401,
+        priority: 2,
+        capture: true,
+        stored_at: new Date(),
+      },
+      { start_id: 501, end_id: 600, next_id: 501, priority: 500 },
+      { start_id: 601, end_id: 700, next_id: 601, priority: 4200, status: 'failed' },
+    ]
+    const ids = await db
+      .table('ranges')
+      .multiInsert(ranges.map((r) => ({ group_id: g.id, ...r })))
+      .returning('id')
+    await db.table('group_leases').multiInsert([
+      {
+        group_id: g.id,
+        lane: 'history',
+        node_id: n.id,
+        leased_at: new Date(),
+        heartbeat_at: new Date(),
+        range_id: ids[3].id,
+        snapshot_next: 301,
+        snapshot_at: new Date(Date.now() - 600_000),
+        ranges_done: 1,
+        headers: 100,
+      },
+      {
+        group_id: g.id,
+        lane: 'tail',
+        node_id: null,
+        heartbeat_at: new Date(Date.now() - 3600_000),
+        snapshot_next: 1,
+        snapshot_at: new Date(Date.now() - 3600_000),
+      },
+    ])
+    await writeFile(join(dir, `${g.id}.gz`), Buffer.alloc(2_000_000))
+    await writeFile(join(dir, `${g.id}-tail.gz`), Buffer.alloc(1000))
+    // A group planned from its low watermark 1000: its ancient range counts in no band and is never leased.
+    const [old] = await db
+      .table('groups')
+      .insert({ name: 'alt.binaries.ancient', priority: 2 })
+      .returning('id')
+    await db.table('ranges').multiInsert([
+      { group_id: old.id, start_id: 1, end_id: 999, next_id: 1, priority: 0, status: 'ancient' },
+      { group_id: old.id, start_id: 1000, end_id: 1099, next_id: 1000, priority: 0 },
+    ])
+
+    const s = await StatusService.collect()
+    assert.deepEqual(
+      s.bands.map((b) => [
+        b.band,
+        b.planned,
+        b.plannedArticles,
+        b.leased,
+        b.done,
+        b.doneArticles,
+        b.failed,
+      ]),
+      [
+        ['tail', 1, 100, 0, 0, 0, 0],
+        ['raised', 1, 100, 0, 0, 0, 0],
+        ['window 0', 0, 0, 1, 1, 100, 0],
+        ['window 2', 1, 100, 0, 0, 0, 0],
+        ['history 0-999', 2, 200, 0, 0, 0, 0],
+        ['history 4000+', 0, 0, 0, 0, 0, 1],
+      ]
+    )
+    assert.deepEqual(s.ancient, { groups: 1, articles: 999 })
+    assert.deepEqual(s.lastHour, {
+      indexedArticles: 100,
+      doneRanges: 1,
+      storedArticles: 100,
+      storedRanges: 1,
+    })
+    assert.deepEqual(s.corpus, { pendingRanges: 1, pendingArticles: 100 })
+    assert.deepEqual(s.spool, { ranges: 1, chunks: 2 })
+    assert.deepEqual(s.tail, { groups: 1, ranges: 1, articles: 100 })
+    assert.lengthOf(s.held, 1)
+    assert.include(s.held[0], {
+      node: 'status-node',
+      group: 'alt.binaries.status',
+      lane: 'history',
+      snapshotNext: 301,
+      snapshotMb: 2,
+      rangesDone: 1,
+      headers: 100,
+    })
+    assert.deepEqual(s.held[0].range, { start: 301, end: 400, next: 351 })
+    assert.isAtLeast(s.held[0].snapshotSeconds!, 599)
+    assert.equal(s.snapshots.count, 2)
+    assert.deepEqual(
+      s.snapshots.largest.map((f) => [f.group, f.lane, f.mb]),
+      [
+        ['alt.binaries.status', 'history', 2],
+        ['alt.binaries.status', 'tail', 0.001],
+      ]
+    )
+    assert.equal(s.releases.provisional, 0)
+    assert.deepEqual(s.probes, { ready: 0, waiting: 0, pendingWaiting: 0, pendingNamed: 0 })
+    assert.equal(s.nodes.find((x) => x.name === 'status-node')?.lastSeenSeconds, 0)
+    assert.isAbove(s.catalog.gb, 0)
+
+    const text = StatusService.render(s)
+    assert.include(text, 'window 0')
+    assert.include(text, 'alt.binaries.status  history  status-node  301-400  351')
+    assert.include(text, 'Snapshots: 2 files')
+    assert.include(text, 'Ancient: 999 articles below the low watermark of 1 groups')
+  })
+})

tests/unit/coordinator/planner_rows.spec.ts

@@ -0,0 +1,36 @@
+import { test } from '@japa/runner'
+import { historyRows } from '#services/planner'
+
+test.group('Planner / history rows', () => {
+  test('a group planned for the first time: the ancient range below the low watermark, then the slices', ({
+    assert,
+  }) => {
+    const rows = historyRows(7, { low: 1001, high: 3500 }, 0, 1000)
+    assert.deepEqual(
+      rows.map((r) => [r.start_id, r.end_id, r.next_id, r.priority, r.status ?? 'planned']),
+      [
+        [1, 1000, 1, 0, 'ancient'],
+        [1001, 2000, 1001, 0, 'planned'],
+        [2001, 3000, 2001, 400, 'planned'],
+        [3001, 3500, 3001, 800, 'planned'],
+      ]
+    )
+  })
+
+  test('no ancient range when the group starts at article 1, nor when planning again', ({
+    assert,
+  }) => {
+    assert.deepEqual(
+      historyRows(7, { low: 1, high: 1500 }, 0, 1000).map((r) => [r.start_id, r.end_id, r.status]),
+      [
+        [1, 1000, undefined],
+        [1001, 1500, undefined],
+      ]
+    )
+    // Planned before up to 3500: only the new articles, no second ancient range.
+    assert.deepEqual(
+      historyRows(7, { low: 1200, high: 4100 }, 3500, 1000).map((r) => [r.start_id, r.end_id]),
+      [[3501, 4100]]
+    )
+  })
+})