diff --git a/REBUILD.md b/REBUILD.md index f93739d..6f9a891 100644 --- a/REBUILD.md +++ b/REBUILD.md @@ -409,6 +409,31 @@ releases, a third of them one poster's image dump (27.7 M rows, LOG.md consecutive ranges of a.b.ath (2016): one range is 15 hours of posting, so without the carry no release would finish inside a range (LOG.md 2026-10-26 (111)). + **The group lease (2026-10-26, the user's question in LOG.md (117)):** + a worker thread leases a *group*, not a range, and asks the coordinator + for the group's next range each time it finished one; the coordinator + keeps one cursor a group (`group_leases`: the holder, its heartbeat, + the next range in article order, the snapshot it stands at). The + answer is the range (id, span, resume point), the rules version and the + stream URL — nothing about where the chunks are: the stream serves a + stored range's chunks from the corpus file and a live one's from the + spool, so a worker needs no mount and no corpus layout, and a remote + worker is the same binary with another URL. The overhead is a copy + through the node API: the corpus is ~30 B an article compressed (2016; + 16–18 for 2021 and 2025), eight local workers at the bench's rate are + ~170 MB/s, which the HDD under the corpus bounds either way; a remote + worker is bounded by its link (100 Mbit ≈ 400k articles/s). History + therefore no longer waits for the corpus fill: a group's stored ranges + are read from disk, an unstored one is fetched live and captured into + the corpus on the way, and the prefetcher runs ahead of each held + group's cursor (plus the capture work left). A group with no planned + range left is the tail: the worker keeps the group and asks again + every minute, the planner's tail ranges fall into its order. A lease + whose holder went quiet for LEASE_TTL_SECONDS is taken over: the new + holder downloads the group's snapshot (5) and redoes the ranges since + it; the batch store's keys make the redo idempotent. The replay + harness (8) stays for offline runs; the driver on homelab goes once the + worker runs. 2. **The window.** The worker reads a block in article order and keeps its open files and releases in memory (counters, names, sizes; the segment @@ -490,8 +515,31 @@ releases, a third of them one poster's image dump (27.7 M rows, LOG.md of writes, so the state is snapshotted at a lease's end and on a timer, to a file on bulk, and a crash redoes the ranges since the last snapshot from the corpus or the prefetch cache: the deterministic keys make the - redo idempotent. Not built yet; the replay harness (8) keeps a group's - window in one process. + redo idempotent. **The snapshot goes through the API** (2026-10-26): + `PUT /node/groups/:id/snapshot` with the serialized window (the + frontier, the open files and releases, the open albums, the upload + numbers, the batch-counter table; gzip) and the article it stands at + (the last batch's `next`), kept by the coordinator on bulk as + `/`; `GET` gives it back to the next holder. A + worker snapshots when it gives a group back (a stop) and every + SNAPSHOT_MINUTES while it holds one, always right after a batch was + stored, so the snapshot's article is a resume point. The replay + harness (8) keeps a group's window in one process without one. + **Built 2026-10-26** (LOG.md (117)): migration 6500 `group_leases`; + `app/services/group_leases.ts` (lease by the most urgent planned + range within LEASE_MIN_PRIORITY..LEASE_MAX_PRIORITY, the longest free + first; takeover after LEASE_TTL_SECONDS, the ranges from the snapshot's + article on planned again; next; heartbeat; release; the snapshot + streamed to `SNAPSHOT_DIR/.gz`); the node API's + `groups/lease`, `groups/:id/next|heartbeat|release|snapshot`; the + range stream serving a stored range's chunks from the corpus; the + prefetcher ahead of held groups' cursors, skipping chunks the corpus + has; `RangeIndexer::for_group/begin_range/snapshot/restore` (bincode, + the files a table, SNAPSHOT_VERSION 1); `edge-node` holding groups by + default (`--ranges` keeps the old path). Tests: the lease order, + takeover and redo, the endpoints with a corpus-served stream and the + snapshot through the API, the snapshot restoring a window between two + ranges (the two-range release one, the album joined). 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 @@ -701,9 +749,12 @@ working on it as it fills; the scratch run's numbers (63 groups) were read before (LOG.md (115)). The tail waits for the window worker. Next: the replay on homelab over the 2016 window (the NVMe bytes per -header for real), the ordered range leases with the carried window in -the edge-node and its snapshot (design 1 and 5), the prober writing -`release_texts` kind 2, the spot/NZB match on pending keys (decision f). Gone from the list: block +header for real), the window worker (design 1 and 5: group leases, the +stream serving stored chunks, the snapshot through the API; being built, +LOG.md (117)), the prober writing `release_texts` kind 2, the spot/NZB +match on pending keys (decision f), the tail's provisional writes at a +range end (the window's open releases shown while they upload: the files +sent provisionally marked, the finish sending the rest). Gone from the list: block leases, the planner and the block merger (design 1, amended). ## Decisions diff --git a/app/controllers/node_controller.ts b/app/controllers/node_controller.ts index cd3dddd..fb50ad7 100644 --- a/app/controllers/node_controller.ts +++ b/app/controllers/node_controller.ts @@ -1,9 +1,12 @@ import type { HttpContext } from '@adonisjs/core/http' 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 BatchStore, { StaleBatch } from '#services/batch_store' import Spool from '#services/spool' +import Corpus from '#services/corpus' import { KEEPALIVE, encodeFrame, fromWire, type WireBatch } from '#lib/edge/index' /** @@ -30,15 +33,82 @@ export default class NodeController { return lease } - /** The range's chunks from its resume point, waiting for those still being fetched. */ + // 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. + + async leaseGroup(ctx: HttpContext) { + const lease = await GroupLeases.lease(this.node(ctx)) + if (!lease) return ctx.response.noContent() + return lease + } + + async nextRange(ctx: HttpContext) { + try { + const next = await GroupLeases.next(Number(ctx.params.id), this.node(ctx)) + if (!next) return ctx.response.noContent() + return next + } catch (error) { + if (error instanceof LeaseLost) return ctx.response.conflict({ error: error.message }) + throw error + } + } + + async groupHeartbeat(ctx: HttpContext) { + return this.lifecycle(ctx, (groupId, nodeId) => GroupLeases.heartbeat(groupId, nodeId)) + } + + async releaseGroup(ctx: HttpContext) { + return this.lifecycle(ctx, (groupId, nodeId) => GroupLeases.release(groupId, nodeId)) + } + + /** + * PUT /groups/:id/snapshot?next=N: the window's serialized state standing at article N, streamed to disk as sent + * (gzip; the body parser leaves an octet-stream alone, so the request is still readable here). + */ + async putSnapshot(ctx: HttpContext) { + 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)) + } + + async getSnapshot(ctx: HttpContext) { + try { + const snapshot = await GroupLeases.getSnapshot(Number(ctx.params.id), this.node(ctx)) + 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)) + ctx.response.header('X-Snapshot-Next', String(snapshot.next)) + return ctx.response.stream(snapshot.stream) + } catch (error) { + if (error instanceof LeaseLost) return ctx.response.conflict({ error: error.message }) + throw error + } + } + + /** + * The range's chunks from its resume point: a stored range's from the corpus (a worker needs no mount; REBUILD.md + * design 1), else from the spool, waiting for those still being fetched. + */ async stream(ctx: HttpContext) { const nodeId = this.node(ctx) const rangeId = Number(ctx.params.id) const range = await this.leased(ctx, rangeId, nodeId) if (!range) return const spans = Spool.chunksOf({ start: Number(range.start_id), end: Number(range.end_id) }, Number(range.next_id)) + const group = range.capture ? await db.from('groups').where('id', range.group_id).select('name').first() : null + const corpusRange = group ? { group: group.name as string, start: Number(range.start_id), end: Number(range.end_id) } : null async function* frames() { for (const span of spans) { + if (corpusRange && (await Corpus.has(corpusRange, span))) { + const data = await Corpus.read(corpusRange, span) + if (data) { + yield encodeFrame({ ...span, data }) + await Leases.heartbeat(rangeId, nodeId) + continue + } + } const waitUntil = Date.now() + CHUNK_WAIT_MS let quietSince = Date.now() while (!(await Spool.has(rangeId, span))) { @@ -75,7 +145,11 @@ export default class NodeController { } async finish(ctx: HttpContext) { - return this.lifecycle(ctx, (rangeId, nodeId) => Leases.finish(rangeId, nodeId)) + 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)) + }) } /** POST /ranges/:id/corrupt { from }: a chunk that doesn't inflate; the range goes back. */ diff --git a/app/services/group_leases.ts b/app/services/group_leases.ts new file mode 100644 index 0000000..05dca47 --- /dev/null +++ b/app/services/group_leases.ts @@ -0,0 +1,164 @@ +import db from '@adonisjs/lucid/services/db' +import { createReadStream, createWriteStream } from 'node:fs' +import { mkdir, rename, stat } from 'node:fs/promises' +import { join } from 'node:path' +import type { Readable } from 'node:stream' +import { pipeline } from 'node:stream/promises' +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' + +/** + * 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. + * + * - 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. + */ +class GroupLeases { + get dir(): string { + return coordinatorConfig.snapshotDir + } + + async lease(nodeId: number): Promise { + 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' : '' + 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))`) + const { rows } = await trx.rawQuery( + `SELECT g.id, g.name, g.priority, l.node_id AS holder, l.snapshot_next, todo.priority AS todo_priority + FROM groups g + JOIN LATERAL (SELECT min(r.priority) AS priority FROM ranges r WHERE r.group_id = g.id AND r.status = 'planned' AND r.next_id <= r.end_id AND r.priority BETWEEN ? AND ? ${history}) todo ON true + 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] + ) + 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. + 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] + ) + } else { + await trx.rawQuery(`UPDATE ranges SET status = 'planned', node_id = NULL WHERE group_id = ? AND status = 'leased'`, [groupId]) + } + await trx.rawQuery( + `INSERT INTO group_leases (group_id, node_id, leased_at, heartbeat_at, range_id) + 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] + ) + await trx.from('nodes').where('id', nodeId).update({ last_seen_at: new Date() }) + return { + group: { id: groupId, name: row.name, priority: Number(row.priority) }, + 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() + if (!lease || Number(lease.node_id) !== nodeId) throw new LeaseLost(`Group ${groupId} 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) + 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 row = + mine ?? + ( + await db.rawQuery( + `UPDATE ranges r SET status = 'leased', node_id = ?, leased_at = now(), heartbeat_at = now(), attempts = attempts + 1, rule_version = ? + WHERE r.id = ( + SELECT id FROM ranges WHERE group_id = ? AND status = 'planned' AND next_id <= end_id AND priority BETWEEN ? AND ? + ORDER BY start_id LIMIT 1 FOR UPDATE SKIP LOCKED) + RETURNING r.id, r.start_id, r.end_id, r.next_id`, + [nodeId, rules.version, groupId, min, max] + ) + ).rows[0] + await db + .from('group_leases') + .where('group_id', groupId) + .update({ range_id: row ? Number(row.id) : null, heartbeat_at: new Date() }) + if (!row) return null + return { range: { id: Number(row.id), start: Number(row.start_id), end: Number(row.end_id), next: Number(row.next_id) } } + } + + 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() }) + 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) { + await db.rawQuery(`UPDATE group_leases SET ranges_done = ranges_done + 1, headers = headers + ?, range_id = NULL WHERE group_id = ?`, [headers, groupId]) + } + + /** The worker gives the group back (after its snapshot): its range stays at its resume point for the next holder. */ + async release(groupId: number, nodeId: number) { + await this.held(groupId, nodeId) + await db.from('ranges').where('group_id', groupId).where('status', 'leased').where('node_id', nodeId).update({ status: 'planned', node_id: null }) + await db.from('group_leases').where('group_id', groupId).update({ node_id: null, range_id: null }) + } + + private snapshotPath(groupId: number) { + return join(this.dir, `${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) + await mkdir(this.dir, { recursive: true }) + const path = this.snapshotPath(groupId) + await pipeline(body, createWriteStream(`${path}.part`)) + await rename(`${path}.part`, path) + await db.from('group_leases').where('group_id', groupId).update({ snapshot_next: next, snapshot_at: new Date(), heartbeat_at: new Date() }) + } + + /** The snapshot's stream and size, or null when there is none. */ + async getSnapshot(groupId: number, nodeId: number): Promise<{ stream: Readable; size: number; next: number } | null> { + const lease = await this.held(groupId, nodeId) + if (lease.snapshot_next === null) return null + const path = this.snapshotPath(groupId) + try { + const { size } = await stat(path) + return { stream: createReadStream(path), size, next: Number(lease.snapshot_next) } + } catch { + return null + } + } +} + +export default new GroupLeases() diff --git a/app/services/prefetcher.ts b/app/services/prefetcher.ts index aa3a8e3..07f93cc 100644 --- a/app/services/prefetcher.ts +++ b/app/services/prefetcher.ts @@ -425,10 +425,18 @@ 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[] = ( + await db + .from('group_leases') + .whereNotNull('node_id') + .where('heartbeat_at', '>=', new Date(Date.now() - coordinatorConfig.leaseTtlSeconds * 1000)) + .select('group_id') + ).map((l) => Number(l.group_id)) // Planned ranges ahead: those started already, then new ones up to SPOOL_AHEAD_PER_LEASE per lease. const ahead = Math.max( coordinatorConfig.spool.minAhead, - coordinatorConfig.spool.aheadPerLease * leased.length + coordinatorConfig.spool.aheadPerLease * Math.max(leased.length, held.length) ) const started = await rows(db.query()) .where('r.status', 'planned') @@ -448,14 +456,17 @@ export class Prefetcher { const want = Math.max(ahead - Number(ready), short) // The catalog at its cap on the NVMe: no history fetched ahead (CatalogSpace). const full = await CatalogSpace.full() + // Ahead of each held group's cursor: its next ranges in article order, a range of every held group before any + // group's second; without held groups (no window worker running) nothing is fetched ahead for workers. const fresh = - room && want > 0 + room && want > 0 && held.length > 0 ? await rows(db.query()) .where('r.status', 'planned') .where('r.spooled_chunks', 0) .whereRaw('r.next_id <= r.end_id') + .whereIn('r.group_id', held) .where((q: any) => (full ? q.where('r.priority', '<', 0) : q)) - .orderBy([{ column: 'r.priority' }, { column: 'r.id' }]) + .orderByRaw('row_number() OVER (PARTITION BY r.group_id ORDER BY r.start_id), r.priority, r.start_id') .limit(want) : [] const targets: Target[] = [] @@ -464,9 +475,12 @@ export class Prefetcher { const end = Number(row.end_id) const total = Spool.chunksOf({ start, end }).length const missing: ChunkSpan[] = [] + // A chunk already in the corpus isn't spooled: the stream serves it from there (REBUILD.md design 1). + const corpusRange = row.capture === true ? { group: row.group as string, start, end } : null for (const span of Spool.chunksOf({ start, end }, Number(row.next_id))) { - if (!this.claimed.has(`${row.id}:${span.from}`) && !(await Spool.has(Number(row.id), span))) - missing.push(span) + if (this.claimed.has(`${row.id}:${span.from}`) || (await Spool.has(Number(row.id), span))) continue + if (corpusRange && (await Corpus.has(corpusRange, span))) continue + missing.push(span) } if (missing.length > 0) { targets.push({ diff --git a/config/coordinator.ts b/config/coordinator.ts index 8541402..f38fa4f 100644 --- a/config/coordinator.ts +++ b/config/coordinator.ts @@ -31,6 +31,13 @@ export default { }, /** A leased range without a heartbeat for this long goes back to the queue. */ leaseTtlSeconds: env.get('LEASE_TTL_SECONDS', 15 * 60), + /** + * The ranges group leases hand out (REBUILD.md design 1), by priority: the tail is negative, the capture windows + * 0–2 (2016, 2021, 2025), the rest of the history above (its position in the group, or 4000+). Bounds so a + * deployment can run one window first (homelab 2026-10-26: 0..0, the 2016 window); unbounded by default. + */ + leasePriorityMin: env.get('LEASE_MIN_PRIORITY', -1_000_000), + leasePriorityMax: env.get('LEASE_MAX_PRIORITY', 1_000_000), /** 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), @@ -146,6 +153,11 @@ export default { */ filter: env.get('CORPUS_FILTER', true), }, + /** + * The group windows' snapshots (REBUILD.md, "Voyager 3: window indexing", 5; app/services/group_leases.ts): a + * file a group on bulk, next to the spool by default. + */ + snapshotDir: env.get('SNAPSHOT_DIR', `${env.get('SPOOL_DIR', app.tmpPath('spool'))}/snapshots`), spool: { dir: env.get('SPOOL_DIR', app.tmpPath('spool')), /** Ranges kept fetched ahead, per active lease. */ diff --git a/database/migrations/1770000006500_group_leases.ts b/database/migrations/1770000006500_group_leases.ts new file mode 100644 index 0000000..19e2c19 --- /dev/null +++ b/database/migrations/1770000006500_group_leases.ts @@ -0,0 +1,33 @@ +import { BaseSchema } from '@adonisjs/lucid/schema' + +/** + * Group leases (REBUILD.md, "Voyager 3: window indexing", 1 and 5; 2026-10-26): a worker thread holds a group, not + * a range, and indexes its ranges in article order through one carried window. One row a group: the holder and + * its heartbeat, the range it works, and the snapshot of its window (the article it stands at: the next one to + * index) that the coordinator keeps on bulk. Ranges stay what is planned, fetched and finished. + */ +export default class extends BaseSchema { + async up() { + this.schema.raw(` + CREATE TABLE group_leases ( + group_id integer PRIMARY KEY REFERENCES groups (id) ON DELETE CASCADE, + -- The holder; null between holders. A holder quiet for LEASE_TTL_SECONDS is taken over. + node_id integer, + leased_at timestamptz, + heartbeat_at timestamptz, + -- The range the holder works (null between ranges). + range_id bigint, + -- The window's snapshot: the article it stands at (a resume point), and when it was taken; null: none. + snapshot_next bigint, + snapshot_at timestamptz, + ranges_done integer NOT NULL DEFAULT 0, + headers bigint NOT NULL DEFAULT 0, + created_at timestamptz NOT NULL DEFAULT now() + )`) + this.schema.raw(`CREATE INDEX group_leases_held_idx ON group_leases (heartbeat_at) WHERE node_id IS NOT NULL`) + } + + async down() { + this.schema.raw(`DROP TABLE IF EXISTS group_leases`) + } +} diff --git a/deploy/compose.yaml b/deploy/compose.yaml index b995add..3dc894a 100644 --- a/deploy/compose.yaml +++ b/deploy/compose.yaml @@ -175,6 +175,9 @@ services: environment: COORDINATOR_URL: http://node-api:3335 NODE_TOKEN: ${NODE_TOKEN} + # 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} stop_grace_period: 60s depends_on: node-api: diff --git a/edge/Cargo.lock b/edge/Cargo.lock index d56c87d..119343e 100644 --- a/edge/Cargo.lock +++ b/edge/Cargo.lock @@ -44,6 +44,15 @@ version = "0.23.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "ac07cdecf99051d9a5238b80f35af32cdeba5b336e55d957b318b50137e18da5" +[[package]] +name = "bincode" +version = "1.3.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b1f45e9417d87227c7a56d22e471c6206462cba514c7590c09aff4cf6d1ddcad" +dependencies = [ + "serde", +] + [[package]] name = "block-buffer" version = "0.10.4" @@ -145,6 +154,7 @@ name = "edge" version = "0.1.0" dependencies = [ "base64 0.22.1", + "bincode", "chrono", "flate2", "libdeflater", diff --git a/edge/Cargo.toml b/edge/Cargo.toml index 849cd6d..e5511c1 100644 --- a/edge/Cargo.toml +++ b/edge/Cargo.toml @@ -14,6 +14,7 @@ path = "src/bin/edge-node.rs" [dependencies] base64 = "0.22" +bincode = "1.3" chrono = "0.4" flate2 = { version = "1", default-features = false, features = ["zlib-rs"] } libdeflater = "1.26.1" diff --git a/edge/README.md b/edge/README.md index 9245ddd..b87b9e4 100644 --- a/edge/README.md +++ b/edge/README.md @@ -234,12 +234,24 @@ them: ```sh cargo build --release -COORDINATOR_URL=http://coordinator:3334 NODE_TOKEN=… ./target/release/edge-node [--threads N] [--batch-articles 500000] [--ranges N] [--max-mbit M] +COORDINATOR_URL=http://coordinator:3334 NODE_TOKEN=… ./target/release/edge-node [--threads N] [--batch-articles 500000] [--snapshot-minutes 30] [--groups N] [--max-mbit M] ``` -- **One range per thread** (default one per core): one thread reads the - coordinator's stream and inflates a frame ahead, another posts the - batches in order (one may wait), the indexer goes on meanwhile. +- **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 + groups/:id/next`), streams each range's chunks (`GET ranges/:id/stream`: + the corpus or the spool, the worker can't tell) through one carried + window, posts a batch at least at every range end and finishes the + range; one thread reads the stream and inflates a frame ahead, another + posts the batches in order, the indexer goes on meanwhile. The window's + snapshot (bincode, gzip) goes to the coordinator every + `--snapshot-minutes` (EDGE_SNAPSHOT_MINUTES, 30) right after a batch, + and when the group is given back: caught up (`next` answers 204: the + tail), a stop, a lost lease or a failure; the next holder restores it + (`GET groups/:id/snapshot`) and continues at its article. `--groups N` + stops after N groups; `--ranges N` is the old per-range lease + (`POST lease`), one range per thread without a carried window. - **As `commands/edge_node.ts`:** retries for network trouble and 5xx; 409 means the lease is lost (the range stops, isn't given back); a batch once 500k articles or 100k files wait (a picture group's batches diff --git a/edge/src/bin/edge-node.rs b/edge/src/bin/edge-node.rs index 1ba9bd1..3b07e66 100644 --- a/edge/src/bin/edge-node.rs +++ b/edge/src/bin/edge-node.rs @@ -1,16 +1,20 @@ -//! The edge worker in Rust (REBUILD.md, part 2; commands/edge_node.ts): leases -//! ranges from the coordinator, reads their XOVER chunks from its stream, -//! indexes them (the same RangeIndexer as edge-bench, kept identical to -//! the TypeScript by check.ts) and posts the batches. One range per thread. +//! The window worker in Rust (REBUILD.md, "Voyager 3: window indexing", 1 and 5): each thread leases a *group* +//! from the coordinator, takes its ranges in article order, reads each range's XOVER chunks from the coordinator's +//! stream (the corpus or the spool, the worker can't tell), indexes them through one carried window and posts the +//! batches, a batch at least at every range end; the window's snapshot goes to the coordinator every +//! --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] [--ranges N] [--max-mbit M] +//! COORDINATOR_URL=… NODE_TOKEN=… edge-node [--threads N] [--batch-articles 500000] [--snapshot-minutes 30] [--groups N] [--max-mbit M] //! -//! SIGTERM/SIGINT: each thread gives its range back after the current chunk -//! (the next worker resumes at the last stored batch). +//! 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 +//! coordinator without group leases). use edge::indexer::{Batch, RangeIndexer}; -use edge::lease::LeaseJson; +use edge::lease::{GroupLeaseJson, LeaseJson, NextRangeJson, RangeSpanJson}; use edge::wire; +use flate2::read::GzDecoder; use flate2::write::GzEncoder; use serde::Deserialize; use std::collections::HashMap; @@ -163,6 +167,81 @@ impl Api { } } + // Group leases (REBUILD.md design 1 and 5). + + fn lease_group(&self) -> Result, ApiError> { + let mut response = self.request("POST", "groups/lease", None, false, "application/json")?; + match response.status().as_u16() { + 204 => Ok(None), + 200 => { + let text = response.body_mut().read_to_string().map_err(|e| ApiError::Other(format!("groups/lease: {e}")))?; + serde_json::from_str(&text).map(Some).map_err(|e| ApiError::Other(format!("groups/lease: {e}"))) + } + status => Err(ApiError::Other(format!("groups/lease: HTTP {status}"))), + } + } + + fn next_range(&self, group: i64) -> Result, ApiError> { + let mut response = self.request("POST", &format!("groups/{group}/next"), None, false, "application/json")?; + match response.status().as_u16() { + 204 => Ok(None), + 200 => { + let text = response.body_mut().read_to_string().map_err(|e| ApiError::Other(format!("next: {e}")))?; + serde_json::from_str::(&text).map(|n| Some(n.range)).map_err(|e| ApiError::Other(format!("next: {e}"))) + } + status => Err(ApiError::Other(format!("next: HTTP {status}"))), + } + } + + fn group_call(&self, group: i64, action: &str) -> Result<(), ApiError> { + let response = self.request("POST", &format!("groups/{group}/{action}"), None, false, "application/json")?; + match response.status().as_u16() { + 200..=299 => Ok(()), + status => Err(ApiError::Other(format!("{action}: HTTP {status}"))), + } + } + + /// The window's snapshot (bincode), gzipped, standing at `next`. + fn put_snapshot(&self, group: i64, 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 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[..]); + match result { + Ok(response) => match response.status().as_u16() { + 200..=299 => return Ok(body.len()), + 409 => return Err(ApiError::LeaseLost("snapshot: the group is no longer this node's".into())), + status if status < 500 => return Err(ApiError::Other(format!("snapshot: HTTP {status}"))), + status if attempt == RETRIES => return Err(ApiError::Other(format!("snapshot: HTTP {status}"))), + _ => {} + }, + Err(error) if attempt == RETRIES => return Err(ApiError::Other(format!("snapshot: {error}"))), + Err(_) => {} + } + std::thread::sleep(Duration::from_millis(60_000.min(1000 * 2u64.pow(attempt)))); + } + unreachable!() + } + + /// 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")?; + match response.status().as_u16() { + 404 => Ok(None), + 200 => { + let next: i64 = response.headers().get("x-snapshot-next").and_then(|v| v.to_str().ok()).and_then(|v| v.parse().ok()).ok_or_else(|| ApiError::Other("snapshot: no X-Snapshot-Next".into()))?; + let mut gz = GzDecoder::new(response.into_body().into_reader()); + let mut bytes = Vec::new(); + gz.read_to_end(&mut bytes).map_err(|e| ApiError::Other(format!("snapshot: {e}")))?; + Ok(Some((next, bytes))) + } + status => Err(ApiError::Other(format!("snapshot: HTTP {status}"))), + } + } + fn corrupt(&self, range: i64, from: i64) -> Result<(), ApiError> { let body = format!("{{\"from\":{from}}}"); let response = self.request("POST", &format!("ranges/{range}/corrupt"), Some(body.as_bytes()), false, "application/json")?; @@ -275,6 +354,8 @@ fn inflate(raw: &[u8], d: &mut libdeflater::Decompressor) -> Result, Str struct Config { batch_articles: i64, bytes_per_second: Option, + /// Between two snapshots of a held group's window. + snapshot_every: Duration, /// Ranges whose batch was too large (413), and how much smaller their batches are now (halved each time). smaller: Mutex>, } @@ -452,6 +533,255 @@ fn run_range(api: &Arc, lease: LeaseJson, config: &Config, stopping: &Atomi } } +/// What one range of a held group came to. +enum RangeOutcome { + Done, + /// The group is given back (a stop, a lost lease, a failure): no more ranges. + Stop, +} + +/// 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. +#[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 { + let started = Instant::now(); + indexer.begin_range(range.next, range.end); + let post_error: Arc>> = Arc::new(Mutex::new(None)); + let last_contact = Arc::new(AtomicI64::new(0)); + let since = Instant::now(); + let (post_tx, post_rx) = sync_channel::(1); + let poster = { + let (api, totals, post_error, last_contact) = (api.clone(), totals.clone(), post_error.clone(), last_contact.clone()); + std::thread::spawn(move || { + for batch in post_rx { + if post_error.lock().unwrap().is_some() { + continue; + } + let at = Instant::now(); + match api.batch(range.id, &batch) { + Ok((releases, candidates, bytes)) => { + let mut t = totals.lock().unwrap(); + t.post_ms += at.elapsed().as_millis(); + t.bytes += bytes; + t.headers += batch.headers; + t.releases += releases; + t.candidates += candidates; + last_contact.store(since.elapsed().as_millis() as i64, Ordering::Relaxed); + } + Err(e) => *post_error.lock().unwrap() = Some(e), + } + } + }) + }; + let frames = match api.stream(range.id) { + Ok(stream) => read_frames(stream, config.bytes_per_second), + Err(e) => { + log("error", &format!("{group_name} {}–{}: stream failed: {e}; the group goes back.", range.start, range.end)); + drop(post_tx); + let _ = poster.join(); + return RangeOutcome::Stop; + } + }; + let (batch_articles, max_pending) = config.limits(range.id); + let mut batch_from = range.next; + let mut reached_end = false; + let mut corrupt: Option<(i64, i64, String)> = None; + let mut failure: Option = None; + let headers_before = totals.lock().unwrap().headers; + for frame in frames.iter() { + if stopping.load(Ordering::Relaxed) || post_error.lock().unwrap().is_some() { + break; + } + 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"); + } + match frame { + Frame::KeepAlive => continue, + Frame::End => break, + Frame::Failed(e) => { + failure = Some(e); + break; + } + Frame::Chunk { from, to, text } => { + let data = match text { + Ok(data) => data, + Err(e) => { + corrupt = Some((from, to, e)); + break; + } + }; + indexer.add_text(&String::from_utf8_lossy(&data), to + 1); + reached_end = to >= range.end; + // The window carries on past the range's end: no finish here; a batch at the end at least. + if reached_end || to + 1 - batch_from >= batch_articles || indexer.pending() >= max_pending { + if post_tx.send(indexer.take()).is_err() { + break; + } + batch_from = to + 1; + } + if reached_end { + break; + } + } + } + } + // Stopped mid-range: the batch so far goes out, so the snapshot stands at a stored resume point. + if !reached_end && corrupt.is_none() && failure.is_none() && post_error.lock().unwrap().is_none() && indexer.next() > batch_from { + let _ = post_tx.send(indexer.take()); + } + drop(post_tx); + let _ = poster.join(); + let post_error = post_error.lock().unwrap().take(); + if let Some((from, to, e)) = corrupt { + log("warn", &format!("{group_name} {}–{}: chunk {from}–{to} doesn't inflate ({e}); reported, the group goes back", range.start, range.end)); + let _ = api.corrupt(range.id, from); + return RangeOutcome::Stop; + } + match post_error { + Some(ApiError::LeaseLost(m)) => { + log("warn", &format!("{group_name} {}–{}: {m}", range.start, range.end)); + return RangeOutcome::Stop; + } + Some(ApiError::TooLarge(m)) => { + let divisor = { + let mut smaller = config.smaller.lock().unwrap(); + let d = smaller.entry(range.id).or_insert(1); + *d = (*d * 2).min(64); + *d + }; + log("warn", &format!("{group_name} {}–{}: {m}; the group goes back, its batches 1/{divisor} the size from now on.", range.start, range.end)); + return RangeOutcome::Stop; + } + Some(ApiError::Other(m)) => { + log("error", &format!("{group_name} {}–{}: {m}; the group goes back.", range.start, range.end)); + return RangeOutcome::Stop; + } + None => {} + } + if let Some(e) = failure { + log("error", &format!("{group_name} {}–{}: {e}; the group goes back.", range.start, range.end)); + return RangeOutcome::Stop; + } + if !reached_end { + return RangeOutcome::Stop; + } + match api.call(range.id, "finish") { + Ok(()) => { + let t = totals.lock().unwrap(); + let seconds = started.elapsed().as_secs_f64(); + let headers = t.headers - headers_before; + log( + "info", + &format!( + "{group_name} {}–{}: {} headers in {:.0} s ({:.0}/s); {} releases and {} files open in the window.", + range.start, + range.end, + headers, + seconds, + headers as f64 / seconds.max(0.001), + indexer.open_releases(), + indexer.open_files() + ), + ); + RangeOutcome::Done + } + Err(e) => { + log("warn", &format!("{group_name} {}–{}: finish: {e}", range.start, range.end)); + RangeOutcome::Stop + } + } +} + +/// 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) { + match indexer.snapshot() { + Ok(bytes) => match api.put_snapshot(group_id, 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}")), + }, + Err(e) => log("warn", &format!("{group_name}: no snapshot: {e}")), + } +} + +/// 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) { + let group_id = lease.group.id; + let group_name = lease.group.name.clone(); + let rules = lease.rules.rules(); + let mut indexer = match lease.snapshot { + Some(s) => match api.get_snapshot(group_id) { + 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)); + ix + } + 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) + } + }, + 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) + } + Err(e) => { + log("warn", &format!("{group_name}: snapshot: {e}; the group goes back.")); + let _ = api.group_call(group_id, "release"); + return; + } + }, + None => RangeIndexer::for_group(&group_name, rules.clone(), 5), + }; + let totals = Arc::new(Mutex::new(Totals::default())); + let started = Instant::now(); + let mut snapshotted = Instant::now(); + let mut ranges = 0; + loop { + if stopping.load(Ordering::Relaxed) { + break; + } + let range = match api.next_range(group_id) { + Ok(Some(range)) => range, + Ok(None) => { + log("info", &format!("{group_name}: caught up after {ranges} ranges; the group goes back.")); + break; + } + Err(e) => { + log("warn", &format!("{group_name}: next range: {e}; the group goes back.")); + break; + } + }; + 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) { + RangeOutcome::Done => ranges += 1, + RangeOutcome::Stop => break, + } + if snapshotted.elapsed() >= config.snapshot_every { + snapshot(api, group_id, &group_name, &indexer); + snapshotted = Instant::now(); + } + } + snapshot(api, group_id, &group_name, &indexer); + let _ = api.group_call(group_id, "release"); + let t = totals.lock().unwrap(); + log( + "info", + &format!( + "{group_name}: {ranges} ranges, {} headers in {:.0} s, {} release chunks, {} probe candidates; posted {:.1} MB in {:.1} s.", + t.headers, + started.elapsed().as_secs_f64(), + t.releases, + t.candidates, + t.bytes as f64 / 1e6, + t.post_ms as f64 / 1000.0 + ), + ); +} + fn main() { let args: Vec = std::env::args().collect(); let flag = |name: &str| args.iter().position(|a| a == name).and_then(|i| args.get(i + 1)).cloned(); @@ -466,6 +796,7 @@ fn main() { 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), + snapshot_every: Duration::from_secs(60 * flag("--snapshot-minutes").or_else(|| std::env::var("EDGE_SNAPSHOT_MINUTES").ok()).map(|v| v.parse::().unwrap()).unwrap_or(30)), // The node's cap, shared by its threads. bytes_per_second: max_mbit.map(|m| m * 1e6 / 8.0 / threads as f64), smaller: Mutex::new(HashMap::new()), @@ -475,21 +806,39 @@ fn main() { 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())); + let groups: Option = flag("--groups").map(|v| v.parse().unwrap()); + let range_leases = ranges.is_some(); let workers: Vec<_> = (0..threads) .map(|_| { let (api, config, stopping) = (api.clone(), config.clone(), stopping.clone()); std::thread::spawn(move || { let mut done = 0; - while !stopping.load(Ordering::Relaxed) && ranges.is_none_or(|n| done < n) { - match api.lease() { + if range_leases { + while !stopping.load(Ordering::Relaxed) && ranges.is_none_or(|n| done < n) { + match api.lease() { + Ok(Some(lease)) => { + run_range(&api, lease, &config, &stopping); + done += 1; + } + Ok(None) => break, + Err(e) => { + log("warn", &format!("Lease failed: {e}")); + idle(&stopping); + } + } + } + return; + } + while !stopping.load(Ordering::Relaxed) && groups.is_none_or(|n| done < n) { + match api.lease_group() { Ok(Some(lease)) => { - run_range(&api, lease, &config, &stopping); + run_group(&api, lease, &config, &stopping); done += 1; } - Ok(None) if ranges.is_some() => break, + Ok(None) if groups.is_some() => break, Ok(None) => idle(&stopping), Err(e) => { - log("warn", &format!("Lease failed: {e}")); + log("warn", &format!("Group lease failed: {e}")); idle(&stopping); } } diff --git a/edge/src/indexer.rs b/edge/src/indexer.rs index 53073f8..a51c700 100644 --- a/edge/src/indexer.rs +++ b/edge/src/indexer.rs @@ -7,6 +7,7 @@ use crate::codec::{encode_chunk, ChunkFile}; use crate::js::{self, sha1_hex}; use crate::names::{self, is_random_file, JudgedFile, Release, Rules, Verdict}; use rustc_hash::{FxHashMap, FxHashSet}; +use serde::{Deserialize, Serialize}; use std::rc::Rc; pub static mut PROF: [u64; 8] = [0; 8]; @@ -503,7 +504,7 @@ fn median(dates: &mut [i64]) -> i64 { /// Where the block is in posting time: the median Date of the last FRONTIER_ARTICLES plausible ones. Date headers /// alone aren't trusted: a date more than PLAUSIBLE_MS from the frontier is set aside, and only FRONTIER_ARTICLES /// such dates in a row, agreeing among themselves, move the frontier there. -#[derive(Default)] +#[derive(Default, Serialize, Deserialize)] pub struct Frontier { ring: Vec, at: usize, @@ -592,6 +593,163 @@ struct OpenFile { latest: Option, } +// The snapshot (REBUILD.md, "Voyager 3: window indexing", 5): the window's state between two batches, serialized +// (bincode, gzip by the worker) and kept by the coordinator, restored by the group's next holder. Segments share +// their file's FileMeta through an Rc; in the snapshot the files are a table and a segment points at its index. +#[derive(Serialize, Deserialize)] +struct SegSnap { + file: u32, + number: u32, + message_id: String, + bytes: i64, + article: i64, + posted_at: Option, +} + +#[derive(Serialize, Deserialize)] +struct OpenFileSnap { + parts: Vec, + total: u32, + touched: u64, + earliest: Option, + latest: Option, +} + +#[derive(Serialize, Deserialize)] +struct OpenReleaseSnap { + upload: usize, + files: Vec>, + first: Option, + last: Option, + touched: u64, +} + +#[derive(Serialize, Deserialize)] +struct OpenAlbumSnap { + upload: usize, + members: Vec<(String, OpenReleaseSnap)>, + first: Option, + last: Option, + touched: u64, +} + +/// A file's details in the snapshot (FileMeta, its release identity's source as text: a &'static str in memory). +#[derive(Serialize, Deserialize)] +struct FileMetaSnap { + file_key: String, + name: String, + poster: String, + subject: String, + total_parts: u32, + groups: Vec, + release_key: String, + release_name: String, + release_source: String, + release_file_total: Option, +} + +/// The sources release_identity gives (lib/release/source.ts), as the static strings the code compares. +fn source_static(source: &str) -> &'static str { + match source { + "subject" => "subject", + "post" => "post", + "stem" => "stem", + "album" => "album", + "weak-stem" => "weak-stem", + _ => "stem", + } +} + +#[derive(Serialize, Deserialize)] +pub struct Snapshot { + pub version: u32, + pub group: String, + /// The article the window stands at: the next one to index (a batch boundary). + pub next: i64, + frontier: Frontier, + files: Vec, + aggregator: Vec<(String, Vec)>, + aggregator_batch: u64, + open: Vec<(String, Vec)>, + albums: Vec<(String, Vec)>, + uploads: Vec<(String, usize)>, + batch_totals: Vec<(String, i64, u32, u64)>, +} + +/// The snapshot's format version: raised when the state's shape changes (an older snapshot is refused: the group +/// continues at its resume point, one seam). +pub const SNAPSHOT_VERSION: u32 = 1; + +/// The files of a snapshot as they are taken: each FileMeta once, by its Rc. +#[derive(Default)] +struct FileTable { + index: FxHashMap<*const FileMeta, u32>, + files: Vec>, +} + +impl FileTable { + fn seg(&mut self, s: &Seg) -> SegSnap { + let ptr = Rc::as_ptr(&s.file); + let file = match self.index.get(&ptr) { + Some(&i) => i, + None => { + let i = self.files.len() as u32; + self.index.insert(ptr, i); + self.files.push(s.file.clone()); + i + } + }; + SegSnap { file, number: s.number, message_id: s.message_id.clone(), bytes: s.bytes, article: s.article, posted_at: s.posted_at } + } + + fn segs(&mut self, segs: &[Seg]) -> Vec { + segs.iter().map(|s| self.seg(s)).collect() + } + + fn release(&mut self, u: &OpenRelease) -> OpenReleaseSnap { + OpenReleaseSnap { upload: u.upload, files: u.files.iter().map(|f| self.segs(f)).collect(), first: u.first, last: u.last, touched: u.touched } + } + + /// The table's files as the snapshot stores them. + fn into_files(self) -> Vec { + self.files + .iter() + .map(|f| FileMetaSnap { + file_key: f.file_key.clone(), + name: f.name.clone(), + poster: f.poster.clone(), + subject: f.subject.clone(), + total_parts: f.total_parts, + groups: f.groups.clone(), + release_key: f.release.key.clone(), + release_name: f.release.name.clone(), + release_source: f.release.source.to_string(), + release_file_total: f.release.file_total, + }) + .collect() + } +} + +fn file_of(f: FileMetaSnap) -> FileMeta { + FileMeta { + file_key: f.file_key, + name: f.name, + poster: f.poster, + subject: f.subject, + total_parts: f.total_parts, + groups: f.groups, + release: Release { key: f.release_key, name: f.release_name, source: source_static(&f.release_source), file_total: f.release_file_total }, + } +} + +fn seg_of(s: SegSnap, files: &[Rc]) -> Seg { + Seg { file: files[s.file as usize].clone(), number: s.number, message_id: s.message_id, bytes: s.bytes, article: s.article, posted_at: s.posted_at } +} + +fn release_of(u: OpenReleaseSnap, files: &[Rc]) -> OpenRelease { + OpenRelease { upload: u.upload, files: u.files.into_iter().map(|f| f.into_iter().map(|s| seg_of(s, files)).collect()).collect(), first: u.first, last: u.last, touched: u.touched } +} + impl OpenFile { fn distance(&self, t: Option) -> i64 { match (t, self.earliest, self.latest) { @@ -873,6 +1031,107 @@ impl RangeIndexer { } } + /// A worker's indexer for a group it holds (REBUILD.md design 1): ranges come one after another (begin_range), + /// the window carried between them; the batch starts at the first range's resume point. + pub fn for_group(group: &str, rules: Rules, idle: u64) -> Self { + Self::new(group, 0, 0, 0, rules, idle) + } + + /// The next range of the group: the batch starts at its resume point (after a gap too), the range's end is known. + pub fn begin_range(&mut self, next: i64, end: i64) { + self.end = end; + if self.batch.headers == 0 && self.batch.releases.is_empty() && self.batch.candidates.is_empty() { + self.batch.from = next; + self.batch.next = next; + } + } + + /// The article the window stands at: the next one to index (right after a take, a batch boundary). + pub fn next(&self) -> i64 { + self.batch.next + } + + /// The window's state between two batches (right after a take; the batch must be empty), serialized (bincode). + pub fn snapshot(&self) -> Result, String> { + if self.batch.headers != 0 || !self.batch.releases.is_empty() || !self.batch.candidates.is_empty() { + return Err("a snapshot is taken right after a batch, not with one open".to_string()); + } + let mut table = FileTable::default(); + let aggregator: Vec<(String, Vec)> = self + .aggregator + .open + .iter() + .map(|(key, copies)| { + (key.clone(), copies.iter().map(|c| OpenFileSnap { parts: table.segs(&c.parts), total: c.total, touched: c.touched, earliest: c.earliest, latest: c.latest }).collect()) + }) + .collect(); + let open: Vec<(String, Vec)> = self.open.iter().map(|(key, uploads)| (key.clone(), uploads.iter().map(|u| table.release(u)).collect())).collect(); + let albums: Vec<(String, Vec)> = self + .albums + .iter() + .map(|(key, list)| { + ( + key.clone(), + list.iter() + .map(|a| OpenAlbumSnap { upload: a.upload, members: a.members.iter().map(|(k, u)| (k.clone(), table.release(u))).collect(), first: a.first, last: a.last, touched: a.touched }) + .collect(), + ) + }) + .collect(); + let snapshot = Snapshot { + version: SNAPSHOT_VERSION, + group: self.group.clone(), + next: self.batch.next, + frontier: Frontier { ring: self.frontier.ring.clone(), at: self.frontier.at, value: self.frontier.value, start: self.frontier.start, aside: self.frontier.aside.clone(), pending: self.frontier.pending }, + files: table.into_files(), + aggregator, + aggregator_batch: self.aggregator.batch, + open, + albums, + uploads: self.uploads.iter().map(|(k, &n)| (k.clone(), n)).collect(), + batch_totals: self.batch_totals.iter().map(|((poster, total), (n, touched))| (poster.clone(), *total, *n, *touched)).collect(), + }; + bincode::serialize(&snapshot).map_err(|e| e.to_string()) + } + + /// A worker's indexer for a group from its snapshot (the rules fresh from the lease). + pub fn restore(bytes: &[u8], rules: Rules, idle: u64) -> Result { + let snapshot: Snapshot = bincode::deserialize(bytes).map_err(|e| format!("snapshot: {e}"))?; + if snapshot.version != SNAPSHOT_VERSION { + return Err(format!("snapshot version {} (this build writes {SNAPSHOT_VERSION})", snapshot.version)); + } + let files: Vec> = snapshot.files.into_iter().map(|f| Rc::new(file_of(f))).collect(); + let mut ix = Self::for_group(&snapshot.group, rules, idle); + ix.frontier = snapshot.frontier; + ix.aggregator.batch = snapshot.aggregator_batch; + for (key, copies) in snapshot.aggregator { + ix.aggregator.open.insert( + key, + copies + .into_iter() + .map(|c| { + let parts: Vec = c.parts.into_iter().map(|s| seg_of(s, &files)).collect(); + let seen = parts.iter().map(|s| s.number).collect(); + OpenFile { parts, seen, total: c.total, touched: c.touched, earliest: c.earliest, latest: c.latest } + }) + .collect(), + ); + } + for (key, uploads) in snapshot.open { + ix.open.insert(key, uploads.into_iter().map(|u| release_of(u, &files)).collect()); + } + for (key, list) in snapshot.albums { + ix.albums.insert( + key, + list.into_iter().map(|a| OpenAlbum { upload: a.upload, members: a.members.into_iter().map(|(k, u)| (k, release_of(u, &files))).collect(), first: a.first, last: a.last, touched: a.touched }).collect(), + ); + } + ix.uploads = snapshot.uploads.into_iter().collect(); + ix.batch_totals = snapshot.batch_totals.into_iter().map(|(poster, total, n, touched)| ((poster, total), (n, touched))).collect(); + ix.batch = Batch { from: snapshot.next, next: snapshot.next, ..Default::default() }; + Ok(ix) + } + pub fn add_text(&mut self, text: &str, next: i64) { let mut segments: Vec = Vec::new(); let mut crossposts: Vec = Vec::new(); @@ -1488,6 +1747,80 @@ mod tests { assert_eq!(real.file_total, Some(120)); } + #[test] + fn a_snapshot_restores_the_window_between_two_ranges() { + let rules = || { + let mut priorities = FxHashMap::default(); + priorities.insert(GROUP.to_string(), 1); + Rules::defaults(priorities) + }; + let day = 24 * 60; + // Range 1: a release whose last file comes in range 2, a day later; an album's first member; fillers. + let mut first = vec![ + line(10, r#"Two.Ranges [1/2] - "Two.Ranges.part1.rar" yEnc (1/1)"#, 0), + line(11, r#"[1/1] "Scan_001.pdf" yEnc (1/1)"#, 1), + ]; + first.extend(fillers(100, 100, 2, 1)); + // The second file within the upload gap of the first (3 h), the album's second member within its window. + let mut second = vec![ + line(1000, r#"Two.Ranges [2/2] - "Two.Ranges.part2.rar" yEnc (1/1)"#, 100), + line(1001, r#"[1/1] "Scan_002.pdf" yEnc (1/1)"#, 101), + ]; + second.extend(fillers(2000, 300, 3 * day, 1)); + let summary = |batches: Vec| { + let mut got: Vec<(String, &str, usize, bool)> = batches.iter().flat_map(|b| b.releases.iter()).filter(|r| !r.name.starts_with("Filler")).map(|r| (r.name.clone(), r.source, r.files, r.finished)).collect(); + got.sort(); + got + }; + // In one go. + let mut whole = RangeIndexer::for_group(GROUP, rules(), 5); + whole.begin_range(1, 999); + whole.add_text(&first.join("\n"), 1000); + let w1 = whole.take(); + whole.begin_range(1000, 1_000_000); + whole.add_text(&second.join("\n"), 1_000_001); + let w2 = whole.take(); + // With a snapshot between the ranges, restored into a fresh indexer. + let mut before = RangeIndexer::for_group(GROUP, rules(), 5); + before.begin_range(1, 999); + before.add_text(&first.join("\n"), 1000); + let s1 = before.take(); + assert_eq!(before.next(), 1000); + let bytes = before.snapshot().unwrap(); + assert!(bytes.len() > 1000); + let mut after = RangeIndexer::restore(&bytes, rules(), 5).unwrap(); + assert_eq!(after.next(), 1000); + assert_eq!(after.open_releases(), before.open_releases()); + assert_eq!(after.open_files(), before.open_files()); + after.begin_range(1000, 1_000_000); + after.add_text(&second.join("\n"), 1_000_001); + let s2 = after.take(); + assert_eq!(summary(vec![s1, s2]), summary(vec![w1, w2])); + let got = summary(vec![after.take()]); + assert!(got.is_empty()); + // The two-range release is one, finished, with both files; the album joined its second member. + let mut a2 = RangeIndexer::for_group(GROUP, rules(), 5); + a2.begin_range(1, 999); + a2.add_text(&first.join("\n"), 1000); + let _ = a2.take(); + let restored = RangeIndexer::restore(&a2.snapshot().unwrap(), rules(), 5).unwrap(); + let _ = restored; + let s2b = { + let mut r = RangeIndexer::restore(&bytes, rules(), 5).unwrap(); + r.begin_range(1000, 1_000_000); + r.add_text(&second.join("\n"), 1_000_001); + r.take() + }; + let two = s2b.releases.iter().find(|r| r.name == "Two Ranges").expect("the two-range release"); + assert!(two.finished && two.files == 2); + assert!(s2b.releases.iter().any(|r| r.source == "album" && r.files == 2)); + // A snapshot with a batch open is refused; one of another version too. + let mut open = RangeIndexer::for_group(GROUP, rules(), 5); + open.begin_range(1, 999); + open.add_text(&first.join("\n"), 1000); + assert!(open.snapshot().is_err()); + } + #[test] fn the_frontier_ignores_implausible_dates_and_follows_a_pause() { let hour = 3_600_000i64; diff --git a/edge/src/lease.rs b/edge/src/lease.rs index bc4a24e..12d7399 100644 --- a/edge/src/lease.rs +++ b/edge/src/lease.rs @@ -11,6 +11,40 @@ pub struct LeaseJson { pub rules: RulesJson, } +/// A group lease (lib/edge/lease.ts GroupLease): the group a worker thread holds, the rules, the window's snapshot if any. +#[derive(Deserialize)] +pub struct GroupLeaseJson { + pub group: GroupJson, + pub rules: RulesJson, + pub snapshot: Option, +} + +#[derive(Deserialize)] +pub struct GroupJson { + pub id: i64, + pub name: String, + pub priority: i64, +} + +#[derive(Deserialize)] +pub struct SnapshotJson { + pub next: i64, +} + +/// The group's next range (lib/edge/lease.ts NextRange). +#[derive(Deserialize)] +pub struct NextRangeJson { + pub range: RangeSpanJson, +} + +#[derive(Deserialize, Clone, Copy)] +pub struct RangeSpanJson { + pub id: i64, + pub start: i64, + pub end: i64, + pub next: i64, +} + #[derive(Deserialize)] pub struct RangeJson { pub id: i64, diff --git a/lib/edge/index.ts b/lib/edge/index.ts index 6a38955..7a6adee 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 } from './lease.js' +export type { Lease, GroupLease, 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 1e86d24..d125a7a 100644 --- a/lib/edge/lease.ts +++ b/lib/edge/lease.ts @@ -5,3 +5,18 @@ export interface Lease { range: { id: number; group: string; start: number; end: number; next: number } rules: Rules } + +/** + * 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). + */ +export interface GroupLease { + group: { id: number; name: string; priority: number } + rules: Rules + snapshot: { next: number } | null +} + +/** The group's next range in article order, leased to the holder (POST groups/:id/next; 204 when caught up). */ +export interface NextRange { + range: { id: number; start: number; end: number; next: number } +} diff --git a/start/env.ts b/start/env.ts index 19e9cbf..38931ee 100644 --- a/start/env.ts +++ b/start/env.ts @@ -74,6 +74,9 @@ export default await Env.create(new URL('../', import.meta.url), { SPOT_STANDALONE_AFTER_HOURS: Env.schema.number.optional(), CORPUS_AHEAD: Env.schema.number.optional(), PROBE_PIPELINE: Env.schema.number.optional(), + SNAPSHOT_DIR: Env.schema.string.optional(), + LEASE_MIN_PRIORITY: Env.schema.number.optional(), + LEASE_MAX_PRIORITY: 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/start/routes.ts b/start/routes.ts index 682d47d..a7a6acc 100644 --- a/start/routes.ts +++ b/start/routes.ts @@ -63,6 +63,13 @@ router router .group(() => { router.post('lease', [controllers.Node, 'lease']) + // Group leases (REBUILD.md, "Voyager 3: window indexing", 1 and 5): the window worker's. + router.post('groups/lease', [controllers.Node, 'leaseGroup']) + router.post('groups/:id/next', [controllers.Node, 'nextRange']) + router.post('groups/:id/heartbeat', [controllers.Node, 'groupHeartbeat']) + router.post('groups/:id/release', [controllers.Node, 'releaseGroup']) + router.put('groups/:id/snapshot', [controllers.Node, 'putSnapshot']) + router.get('groups/:id/snapshot', [controllers.Node, 'getSnapshot']) router.get('ranges/:id/stream', [controllers.Node, 'stream']) router.post('ranges/:id/batches', [controllers.Node, 'batch']) router.post('ranges/:id/heartbeat', [controllers.Node, 'heartbeat']) diff --git a/tests/functional/group_leases.spec.ts b/tests/functional/group_leases.spec.ts new file mode 100644 index 0000000..6b9b8ce --- /dev/null +++ b/tests/functional/group_leases.spec.ts @@ -0,0 +1,145 @@ +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 coordinatorConfig from '#config/coordinator' +import GroupLeases from '#services/group_leases' +import { LeaseLost } from '#services/leases' +import { Readable } from 'node:stream' + +async function node(name: string) { + return Number((await db.table('nodes').insert({ name }).returning('id'))[0].id) +} + +async function group(name: string, priority: number) { + return Number((await db.table('groups').insert({ name, priority }).returning('id'))[0].id) +} + +async function range(groupId: number, start: number, priority: number, extra: Record = {}) { + const [row] = await db + .table('ranges') + .insert({ group_id: groupId, start_id: start, end_id: start + 99, next_id: start, priority, ...extra }) + .returning('id') + return Number(row.id) +} + +const rangeRow = async (id: number) => (await db.from('ranges').where('id', id).firstOrFail()) as any + +test.group('Group leases', (g) => { + let dir = '' + let snapshotDir = '' + g.each.setup(async () => { + const rollback = await testUtils.db().withGlobalTransaction() + dir = await mkdtemp(join(tmpdir(), 'snapshots-')) + snapshotDir = coordinatorConfig.snapshotDir + coordinatorConfig.snapshotDir = dir + return rollback + }) + g.each.teardown(async () => { + coordinatorConfig.snapshotDir = snapshotDir + 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 ({ + assert, + }) => { + const a = await group('alt.binaries.gl.a', 1) + const b = await group('alt.binaries.gl.b', 2) + // Group a: history at priority 0 and 1; group b: a tail range (priority -3) and history. + const a1 = await range(a, 1, 0) + const a2 = await range(a, 101, 0) + const a3 = await range(a, 201, 1) + const b2 = await range(b, 101, 0) + const b1 = await range(b, 1, -3) + const n1 = await node('gl-1') + const n2 = await node('gl-2') + + const first = await GroupLeases.lease(n1) + assert.equal(first?.group.id, b, 'the tail first') + 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.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 }) + // 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) + await db.from('ranges').where('id', b2).update({ status: 'done', next_id: 201 }) + assert.isNull(await GroupLeases.next(b, n1), 'caught up') + + assert.deepEqual([a1, a2, a3].length, 3) + await assert.rejects(() => GroupLeases.next(a, n1), LeaseLost) + }) + + test('a stale holder is taken over and the ranges after the snapshot are read again; a release keeps the resume point', async ({ + assert, + }) => { + const a = await group('alt.binaries.gl.c', 1) + const r1 = await range(a, 1, 0) + const r2 = await range(a, 101, 0) + const r3 = await range(a, 201, 0) + const r4 = await range(a, 301, 0) + const n1 = await node('gl-4') + const n2 = await node('gl-5') + assert.equal((await GroupLeases.lease(n1))?.group.id, a) + // The worker did r1 and r2, snapshotted at 201 (r3's start), then got halfway through r3 and went quiet. + await db.from('ranges').whereIn('id', [r1, r2]).update({ status: 'done', next_id: db.raw('end_id + 1') }) + await GroupLeases.putSnapshot(a, n1, 201, Readable.from([Buffer.from('window at 201')])) + await GroupLeases.next(a, n1) + await db.from('ranges').where('id', r3).update({ next_id: 251 }) + await db.from('group_leases').where('group_id', a).update({ heartbeat_at: new Date(Date.now() - 24 * 3600_000) }) + + const lease = await GroupLeases.lease(n2) + assert.equal(lease?.group.id, a) + assert.deepEqual(lease?.snapshot, { next: 201 }) + // r1 and r2 stay done (before the snapshot); r3 is read again from the snapshot's article, r4 untouched. + assert.include(await rangeRow(r1), { status: 'done' }) + assert.include(await rangeRow(r2), { status: 'done' }) + assert.include(await rangeRow(r3), { status: 'planned', node_id: null }) + assert.equal(Number((await rangeRow(r3)).next_id), 201) + assert.include(await rangeRow(r4), { status: 'planned' }) + const snapshot = await GroupLeases.getSnapshot(a, n2) + assert.equal(snapshot?.next, 201) + assert.equal(snapshot?.size, 'window at 201'.length) + await assert.rejects(() => GroupLeases.next(a, n1), LeaseLost) + + // The new holder works r3, snapshots at 301 and gives the group back: r3 done, r4 waits at its start. + assert.equal((await GroupLeases.next(a, n2))?.range.id, r3) + await db.from('ranges').where('id', r3).update({ status: 'done', next_id: 301 }) + await GroupLeases.putSnapshot(a, n2, 301, Readable.from([Buffer.from('window at 301')])) + assert.equal((await GroupLeases.next(a, n2))?.range.id, r4) + await db.from('ranges').where('id', r4).update({ next_id: 301 }) + await GroupLeases.release(a, n2) + assert.include(await rangeRow(r4), { status: 'planned', node_id: null }) + assert.isNull((await db.from('group_leases').where('group_id', a).firstOrFail()).node_id) + // Leased again at once (free, not stale): the snapshot at 301 covers everything done. + const again = await GroupLeases.lease(n1) + assert.deepEqual(again?.snapshot, { next: 301 }) + assert.include(await rangeRow(r3), { status: 'done' }) + assert.equal((await GroupLeases.next(a, n1))?.range.id, r4) + }) + + test('a snapshot older than a done range makes that range be read again from its start', async ({ assert }) => { + const a = await group('alt.binaries.gl.d', 1) + const r1 = await range(a, 1, 0) + const r2 = await range(a, 101, 0) + const n1 = await node('gl-6') + await GroupLeases.lease(n1) + // 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('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) + }) +}) diff --git a/tests/functional/node_api.spec.ts b/tests/functional/node_api.spec.ts index f5dd984..0aadb87 100644 --- a/tests/functional/node_api.spec.ts +++ b/tests/functional/node_api.spec.ts @@ -7,6 +7,7 @@ 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 coordinatorConfig from '#config/coordinator' import { decodeFrames, fromWire, toWire, RangeIndexer, type Lease } from '#lib/edge/index' import { parseOverviews } from '#lib/nntp/index' @@ -86,6 +87,72 @@ test.group('Node API', (group) => { assert.equal(range.status, 'done') assert.isFalse(await Spool.has(rangeId, { from: 1, to: 100 })) }) + test('a group lease: the next range in order, its stream from the corpus, the snapshot through the API', async ({ client, assert }) => { + const corpus = await mkdtemp(join(tmpdir(), 'corpus-')) + const snapshots = await mkdtemp(join(tmpdir(), 'snapshots-')) + const saved = { corpus: coordinatorConfig.corpus.dir, snapshots: coordinatorConfig.snapshotDir } + coordinatorConfig.corpus.dir = corpus + coordinatorConfig.snapshotDir = snapshots + try { + const [g] = await db.table('groups').insert({ name: GROUP, priority: 1 }).returning('id') + // Two ranges of the group: the first stored in the corpus (no spool), the second not. + const [r1] = await db.table('ranges').insert({ group_id: g.id, start_id: 1, end_id: 200, next_id: 1, capture: true, stored_at: new Date() }).returning('id') + const [r2] = await db.table('ranges').insert({ group_id: g.id, start_id: 201, end_id: 400, next_id: 201 }).returning('id') + const corpusRange = { group: GROUP, start: 1, end: 200 } + await Corpus.write(corpusRange, { from: 1, to: 100 }, chunk(11), 2) + await Corpus.write(corpusRange, { from: 101, to: 200 }, chunk(111), 2) + const node = await Node.create({ name: 'n1' }) + const token = (await Node.accessTokens.create(node)).value!.release() + + const lease = await client.post('/api/v1/node/groups/lease').bearerToken(token) + lease.assertStatus(200) + const groupId = Number(lease.body().group.id) + assert.equal(groupId, Number(g.id)) + assert.isNull(lease.body().snapshot) + ;(await client.post('/api/v1/node/groups/lease').bearerToken(token)).assertStatus(204) + + const next = await client.post(`/api/v1/node/groups/${groupId}/next`).bearerToken(token) + next.assertStatus(200) + assert.deepEqual(next.body().range, { id: Number(r1.id), start: 1, end: 200, next: 1 }) + + // The stored range's chunks come from the corpus, nothing spooled. + const stream = await client.get(`/api/v1/node/ranges/${r1.id}/stream`).bearerToken(token) + stream.assertStatus(200) + const raw = stream.body() as unknown + const body: Buffer = Buffer.isBuffer(raw) ? raw : Buffer.from(stream.text(), 'binary') + const frames = [] + for await (const frame of decodeFrames([body])) frames.push(frame) + assert.deepEqual(frames.map((f) => [f.from, f.to]), [[1, 100], [101, 200]]) + assert.equal(parseOverviews(unzipSync(frames[0].data).toString('utf8')).length, 3) + + // The snapshot, standing at the second range's start. + const putRaw = (url: string, bytes: Buffer) => { + const request = client.put(url).bearerToken(token) + request.request.set('Content-Type', 'application/octet-stream').send(bytes) + return request + } + ;(await putRaw(`/api/v1/node/groups/${groupId}/snapshot`, Buffer.from('window'))).assertStatus(400) + ;(await putRaw(`/api/v1/node/groups/${groupId}/snapshot?next=201`, Buffer.from('window at 201'))).assertStatus(200) + await db.from('ranges').where('id', r1.id).update({ status: 'done', next_id: 201 }) + const got = await client.get(`/api/v1/node/groups/${groupId}/snapshot`).bearerToken(token) + got.assertStatus(200) + assert.equal(got.header('x-snapshot-next'), '201') + const gotRaw = got.body() as unknown + assert.equal((Buffer.isBuffer(gotRaw) ? gotRaw : Buffer.from(got.text(), 'binary')).toString(), 'window at 201') + assert.equal(Number((await client.post(`/api/v1/node/groups/${groupId}/next`).bearerToken(token)).body().range.id), Number(r2.id)) + + ;(await client.post(`/api/v1/node/groups/${groupId}/heartbeat`).bearerToken(token)).assertStatus(200) + ;(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 }) + } finally { + coordinatorConfig.corpus.dir = saved.corpus + coordinatorConfig.snapshotDir = saved.snapshots + await rm(corpus, { recursive: true, force: true }) + await rm(snapshots, { recursive: true, force: true }) + } + }) + test('a chunk that does not inflate: reported, removed from the spool, the range goes back', async ({ client, assert }) => { const [g] = await db.table('groups').insert({ name: GROUP, priority: 1 }).returning('id') const [r] = await db.table('ranges').insert({ group_id: g.id, start_id: 1, end_id: 300, next_id: 1, spooled_chunks: 3 }).returning('id')