diff --git a/.gitea/workflows/deploy.yaml b/.gitea/workflows/deploy.yaml index da0265b..435c9b5 100644 --- a/.gitea/workflows/deploy.yaml +++ b/.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: diff --git a/REBUILD.md b/REBUILD.md index be27605..e7889ca 100644 --- a/REBUILD.md +++ b/REBUILD.md @@ -573,6 +573,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 + (`-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 diff --git a/adonisrc.ts b/adonisrc.ts index 77c1c33..67d301f 100644 --- a/adonisrc.ts +++ b/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, }, diff --git a/app/controllers/node_controller.ts b/app/controllers/node_controller.ts index d217e71..b73ce32 100644 --- a/app/controllers/node_controller.ts +++ b/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))) }) } diff --git a/app/services/group_leases.ts b/app/services/group_leases.ts index 4675e0c..ce568e1 100644 --- a/app/services/group_leases.ts +++ b/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 - * the snapshot its window stands at. + * window. The coordinator keeps one row a group and lane (group_leases): the holder, its heartbeat, the range it + * 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 - * the windows), the longest unheld first; a holder quiet for LEASE_TTL_SECONDS is taken over. Taking a group - * resets what the window's snapshot doesn't cover: the ranges from the snapshot's article on 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 (or the one this node already has), leased to the node; none when the - * group 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), given to the next holder. + * - Two lanes (2026-10-27): the `history` lane works the ranges the bounds hand out (LEASE_MIN/MAX_PRIORITY, + * LEASE_CAPTURE_ONLY: the windows, the history), the `tail` lane the tail planner's ranges (priority at or below + * the tail's). Each lane has its own holder, window and snapshot, so a thread on the tail never wades through a + * group's history first (a group's ranges come in article order, and a held group's history is days of ranges), and + * the tail of a group whose history is held isn't stuck behind it. The one seam a group is where the history lane + * reaches the tail lane's first range (its ranges are done: the history lane is caught up). + * - lease: the free group with the most urgent planned range of the lane (the ranges' priority), and among those a + * mix: the group with the most planned articles and the longest unheld one (never held first) take turns, so a + * 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; `.gz`, the tail's + * `-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 { - return coordinatorConfig.leaseCaptureOnly ? 'AND r.capture' : '' + /** + * 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 + * 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 { + async lease(nodeId: number, lane: Lane = 'history'): Promise { + // 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 - // The catalog at its cap on the NVMe: the tail only (CatalogSpace), as the range leases did. - const history = (await CatalogSpace.full()) ? 'AND r.priority < 0' : '' + const todo = this.laneFilter(lane, true) + const all = this.laneFilter(lane, false) 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 - FROM groups g - JOIN LATERAL (SELECT min(r.priority) AS priority FROM ranges r WHERE r.group_id = g.id AND r.status = 'planned' AND r.next_id <= r.end_id AND r.priority BETWEEN ? AND ? ${history} ${this.captureOnly}) todo ON true - LEFT JOIN group_leases l ON l.group_id = g.id - WHERE g.active AND todo.priority IS NOT NULL - AND (l.node_id IS NULL OR l.heartbeat_at < now() - make_interval(secs => ?)) - ORDER BY todo.priority, l.heartbeat_at NULLS FIRST, g.priority - LIMIT 1`, - [min, max, ttl] + `WITH todo AS ( + SELECT r.group_id, min(r.priority) AS priority, sum(r.end_id - r.next_id + 1) AS articles + FROM ranges r + WHERE (r.status = 'planned' OR (r.status = 'leased' AND r.heartbeat_at < now() - make_interval(secs => ?))) + AND r.next_id <= r.end_id AND ${todo.sql} + GROUP BY r.group_id + ), free AS ( + SELECT g.id, g.name, g.priority AS group_priority, l.snapshot_next, l.snapshot_at, l.heartbeat_at, + 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) - WHERE group_id = ? AND end_id >= ?::bigint AND (status IN ('leased', 'done') OR next_id > ?::bigint)`, - [row.snapshot_next, groupId, row.snapshot_next, row.snapshot_next] + `UPDATE ranges r SET status = 'planned', node_id = NULL, next_id = greatest(r.start_id, ?::bigint) + WHERE r.group_id = ? AND r.end_id >= ?::bigint AND ${all.sql} + 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'`, - [groupId] + `UPDATE ranges r SET status = 'planned', node_id = NULL WHERE r.group_id = ? AND r.status = 'leased' AND ${all.sql}`, + [groupId, ...all.bindings] ) } await trx.rawQuery( - `INSERT INTO group_leases (group_id, node_id, leased_at, heartbeat_at, range_id) - 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`, - [groupId, nodeId] + `INSERT INTO group_leases (group_id, lane, node_id, leased_at, heartbeat_at, range_id) + VALUES (?, ?, ?, now(), now(), NULL) + ON CONFLICT (group_id, lane) DO UPDATE SET node_id = excluded.node_id, leased_at = now(), heartbeat_at = now(), range_id = NULL`, + [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. */ - async held(groupId: number, nodeId: number) { - const lease = await db.from('group_leases').where('group_id', groupId).first() + /** The group's lease row of the lane, when this node holds it. */ + async held(groupId: number, nodeId: number, lane: Lane = 'history') { + 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 { - await this.held(groupId, nodeId) + /** + * The group's next range of the lane in article order (or the one this node has), leased to the node; null when + * the lane is caught up. + */ + async next(groupId: number, nodeId: number, lane: Lane = 'history'): Promise { + await this.held(groupId, nodeId, lane) const rules = await RuleSets.current() - const { leasePriorityMin: min, leasePriorityMax: max } = coordinatorConfig - const mine = await db - .from('ranges') - .where('group_id', groupId) - .where('status', 'leased') - .where('node_id', nodeId) - .orderBy('start_id') - .first() + const todo = this.laneFilter(lane, true) + const all = this.laneFilter(lane, false) + const mine = ( + await db.rawQuery( + `SELECT r.id, r.start_id, r.end_id, r.next_id FROM ranges r + WHERE r.group_id = ? AND r.status = 'leased' AND r.node_id = ? AND ${all.sql} ORDER BY r.start_id LIMIT 1`, + [groupId, nodeId, ...all.bindings] + ) + ).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) { - await this.held(groupId, nodeId) - await db.from('group_leases').where('group_id', groupId).update({ heartbeat_at: new Date() }) + async heartbeat(groupId: number, nodeId: number, lane: Lane = 'history') { + await this.held(groupId, nodeId, lane) + 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). */ - async rangeDone(groupId: number, headers: number) { + /** The lane a range belongs to, by its priority. */ + 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 = ?`, - [headers, groupId] + `UPDATE group_leases SET ranges_done = ranges_done + 1, headers = headers + ?, range_id = NULL WHERE group_id = ? AND lane = ?`, + [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) { - await this.held(groupId, nodeId) - await db - .from('ranges') - .where('group_id', groupId) - .where('status', 'leased') - .where('node_id', nodeId) - .update({ status: 'planned', node_id: null }) + async release(groupId: number, nodeId: number, lane: Lane = 'history') { + await this.held(groupId, nodeId, lane) + const all = this.laneFilter(lane, false) + await db.rawQuery( + `UPDATE ranges r SET status = 'planned', node_id = NULL + WHERE r.group_id = ? AND r.status = 'leased' AND r.node_id = ? AND ${all.sql}`, + [groupId, nodeId, ...all.bindings] + ) await db .from('group_leases') .where('group_id', groupId) + .where('lane', lane) .update({ node_id: null, range_id: null }) } - private snapshotPath(groupId: number) { - return join(this.dir, `${groupId}.gz`) + snapshotPath(groupId: number, lane: Lane = 'history') { + 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) { - await this.held(groupId, nodeId) + async putSnapshot( + 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) } diff --git a/app/services/prefetcher.ts b/app/services/prefetcher.ts index 2d1aedc..dbdb0d9 100644 --- a/app/services/prefetcher.ts +++ b/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. - const held: number[] = ( + // The groups workers hold (group_leases; REBUILD.md design 1), by lane: the prefetcher runs ahead of their + // 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') - ).map((l) => Number(l.group_id)) + .select('group_id', 'lane') + ).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: - // LEASE_MIN/MAX_PRIORITY, LEASE_CAPTURE_ONLY), a range of every held group before any group's second; without - // held groups (no window worker running) nothing is fetched ahead for workers. + // Ahead of each held cursor: the lane's next ranges in article order (the history lane's within what the group + // leases hand out: LEASE_MIN/MAX_PRIORITY, LEASE_CAPTURE_ONLY, and none at the catalog's cap; the tail lane's + // 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) - .whereBetween('r.priority', [leasePriorityMin, leasePriorityMax]) - .where((q: any) => (leaseCaptureOnly ? q.where('r.capture', true) : q)) - .where((q: any) => (full ? q.where('r.priority', '<', 0) : q)) + .where((q: any) => + q + .where((h: any) => + h + .whereIn('r.group_id', heldHistory) + .where('r.priority', '>', tailPriority) + .whereBetween('r.priority', [leasePriorityMin, leasePriorityMax]) + .where((c: any) => (leaseCaptureOnly ? c.where('r.capture', true) : c)) + ) + .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) : [] diff --git a/app/services/status.ts b/app/services/status.ts new file mode 100644 index 0000000..db887d4 --- /dev/null +++ b/app/services/status.ts @@ -0,0 +1,333 @@ +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 + databaseGb: number + 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 } + held: HeldGroup[] + snapshots: { count: number; gb: number; oldestHours: number | null; largest: SnapshotFile[] } + releases: { estimated: number; provisional: number } + 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 { + const tail = coordinatorConfig.tail.priority + const cap = coordinatorConfig.catalogMaxNvmeGb + 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, + pg_database_size(current_database()) AS database_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 by priority, the history by its thousand. + const { rows: bands } = await db.rawQuery( + `SELECT CASE WHEN priority <= ? THEN 'tail' + WHEN priority < 0 THEN 'raised' + WHEN capture 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(priority)`, + [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 + 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() + const files: SnapshotFile[] = [] + const names = new Map( + (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) + + 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 count(*) FROM probe_candidates WHERE status = 'waiting' AND ready) AS ready, + (SELECT count(*) FROM probe_candidates WHERE status = 'waiting') 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, + databaseGb: Number(space[0].database_bytes) / 1e9, + 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) }, + 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)'}; database ${s.catalog.databaseGb.toFixed(1)} GB; ${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` + ) + 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() diff --git a/commands/coordinator_status.ts b/commands/coordinator_status.ts new file mode 100644 index 0000000..e8ece3d --- /dev/null +++ b/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)) + } +} diff --git a/config/coordinator.ts b/config/coordinator.ts index 8ea9530..e459e14 100644 --- a/config/coordinator.ts +++ b/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), diff --git a/database/migrations/1770000006600_group_lease_lanes.ts b/database/migrations/1770000006600_group_lease_lanes.ts new file mode 100644 index 0000000..568c948 --- /dev/null +++ b/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)`) + } +} diff --git a/deploy/compose.yaml b/deploy/compose.yaml index 3dc894a..6289dc3 100644 --- a/deploy/compose.yaml +++ b/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: diff --git a/edge/README.md b/edge/README.md index 54c612d..0ac65ba 100644 --- a/edge/README.md +++ b/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 + (`-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); - it talks to `node-api`. + `edge` image target of the Dockerfile, `--threads $EDGE_THREADS` (8), + 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. diff --git a/edge/src/bin/edge-node.rs b/edge/src/bin/edge-node.rs index 35d76f1..a6afc23 100644 --- a/edge/src/bin/edge-node.rs +++ b/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, ApiError> { - let mut response = self.request("POST", "groups/lease", None, false, "application/json")?; + fn lease_group(&self, lane: &str) -> Result, ApiError> { + 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, ApiError> { - let mut response = self.request("POST", &format!("groups/{group}/next"), None, false, "application/json")?; + fn next_range(&self, group: i64, lane: &str) -> Result, ApiError> { + 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> { - let response = self.request("POST", &format!("groups/{group}/{action}"), None, false, "application/json")?; + fn group_call(&self, group: i64, lane: &str, action: &str) -> Result<(), ApiError> { + 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 { + fn put_snapshot(&self, group: i64, lane: &str, next: i64, snapshot: &[u8]) -> Result { 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)>, ApiError> { - let response = self.request("GET", &format!("groups/{group}/snapshot"), None, false, "application/octet-stream")?; + fn get_snapshot(&self, group: i64, lane: &str) -> Result)>, ApiError> { + 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, group_id: i64, group_name: &str, range: RangeSpanJson, indexer: &mut RangeIndexer, config: &Config, stopping: &AtomicBool, totals: &Arc>) -> RangeOutcome { +fn run_group_range(api: &Arc, group_id: i64, lane: &str, group_name: &str, range: RangeSpanJson, indexer: &mut RangeIndexer, config: &Config, stopping: &AtomicBool, totals: &Arc>) -> RangeOutcome { let started = Instant::now(); indexer.begin_range(range.next, range.end); let post_error: Arc>> = Arc::new(Mutex::new(None)); @@ -597,7 +602,7 @@ fn run_group_range(api: &Arc, 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, 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 -/// caught up or the worker stops; then a snapshot, and the group given back. -fn run_group(api: &Arc, lease: GroupLeaseJson, config: &Config, stopping: &AtomicBool) { +/// One held group (one lane of it): its window restored from the snapshot (or new), its ranges one after another +/// until it is caught up or the worker stops; then a snapshot, and the group given back. +fn run_group(api: &Arc, 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, 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, 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, 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); - let _ = api.group_call(group_id, "release"); + snapshot(api, group_id, lane, &group_name, &indexer); + 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 = 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 = 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 = 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) -> bool { + ranges.is_some() +} + fn idle(stopping: &AtomicBool) { let until = Instant::now() + IDLE; while Instant::now() < until && !stopping.load(Ordering::Relaxed) { diff --git a/lib/edge/index.ts b/lib/edge/index.ts index 7a6adee..c64c053 100644 --- a/lib/edge/index.ts +++ b/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' diff --git a/lib/edge/lease.ts b/lib/edge/lease.ts index d125a7a..57b55d4 100644 --- a/lib/edge/lease.ts +++ b/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 - * snapshot of the group's window when there is one (its next article; the body comes from GET groups/:id/snapshot). + * A group's two cursors (REBUILD.md, "Voyager 3: window indexing", 1): `history` works the ranges the lease bounds + * 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 } diff --git a/start/env.ts b/start/env.ts index 42f7ff7..59a9b7d 100644 --- a/start/env.ts +++ b/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(), diff --git a/tests/e2e/edge_node.spec.ts b/tests/e2e/edge_node.spec.ts new file mode 100644 index 0000000..acf10c8 --- /dev/null +++ b/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) + }) +}) diff --git a/tests/e2e/helpers/edge.ts b/tests/e2e/helpers/edge.ts new file mode 100644 index 0000000..dd500c6 --- /dev/null +++ b/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 { + 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 { + 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, 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 \t${date}\t\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')) +} diff --git a/tests/functional/group_leases.spec.ts b/tests/functional/group_leases.spec.ts index bf9aa52..4cdf67a 100644 --- a/tests/functional/group_leases.spec.ts +++ b/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. - assert.deepEqual((await GroupLeases.next(b, n1))?.range, { id: b1, start: 1, end: 100, next: 1 }) - assert.include(await rangeRow(b1), { status: 'leased', node_id: n1 }) + // Ranges in article order, whatever their priority; the tail range isn't the history lane's. + assert.deepEqual((await GroupLeases.next(b, n2))?.range, { id: b2, start: 101, end: 200, next: 101 }) + 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) - await db.from('ranges').where('id', b1).update({ status: 'done', next_id: 101 }) - assert.equal((await GroupLeases.next(b, n1))?.range.id, b2) + assert.equal((await GroupLeases.next(b, n2))?.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') }) }) diff --git a/tests/functional/node_api.spec.ts b/tests/functional/node_api.spec.ts index 8992634..f5d4009 100644 --- a/tests/functional/node_api.spec.ts +++ b/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 diff --git a/tests/functional/status.spec.ts b/tests/functional/status.spec.ts new file mode 100644 index 0000000..888d550 --- /dev/null +++ b/tests/functional/status.spec.ts @@ -0,0 +1,157 @@ +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)) + + 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', 1, 100, 0, 0, 0, 0], + ['history 4000+', 0, 0, 0, 0, 0, 1], + ] + ) + 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') + }) +})