diff --git a/Dockerfile b/Dockerfile index a2490d8..7e2236c 100644 --- a/Dockerfile +++ b/Dockerfile @@ -22,6 +22,8 @@ FROM node:${NODE_VERSION}-bookworm-slim AS app ENV NODE_ENV=production WORKDIR /app COPY --from=app-build /src/build ./ +# The Rust indexer's bench: `node ace replay:corpus` runs a group's corpus through it (the replay harness). +COPY --from=edge-build /usr/local/bin/edge-bench /usr/local/bin/edge-bench USER node # Every service picks its own command, e.g. `node ace coordinator:run`. CMD ["node", "ace", "list"] diff --git a/REBUILD.md b/REBUILD.md index dd8d9f0..2c3c4cf 100644 --- a/REBUILD.md +++ b/REBUILD.md @@ -390,13 +390,25 @@ little in a backfill. That churn is most of the WAL (0.6 TB a day) and the releases, a third of them one poster's image dump (27.7 M rows, LOG.md (102)). -1. **Blocks.** A group's history is still fetched as ranges of 2 M - articles in chunks of 100 k (the spool, the corpus, the prefetcher stay), - but it is *leased* as blocks: consecutive ranges covering at least 30 - days of posting time (`ranges.posted_low/high`), so the window below is - small against a block (≤ 3% of a block's releases touch an edge). A - small group is one block; boneless has a block a week or so. Blocks of - one group run in parallel on different workers, in any order. +1. **One cursor a group (amended 2026-10-26; was: blocks).** A group's + history is still fetched as ranges of 2 M articles in chunks of 100 k + (the spool, the corpus, the prefetcher stay) and leased as ranges, but + a group's ranges are indexed in article order by one worker at a time, + and the window below is *carried* from range to range: the next range + finishes what the previous one left open. The first design leased + blocks of 30 days of posting time, ran a group's blocks in parallel and + joined their edges afterwards (a block merger). The user asked why + (LOG.md 2026-10-25 (110)): downloading stays parallel through the + prefetcher and a Rust thread parses faster than a connection delivers, + so one worker a group loses nothing but the end of a reset, when the + last big group runs alone. The merger, the block cutting and the + `blocks` lease go. Costs: one seam a group where the history meets the + tail (the releases in flight then end as two incomplete halves; history + runs forward in one pass a group, priority only between groups), and + the carried state must survive a restart (5 below). Measured on six + 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)). 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 @@ -412,15 +424,16 @@ releases, a third of them one poster's image dump (27.7 M rows, LOG.md decides later for those without a counter), never updated by the header pass again. Incomplete releases are kept (user, 2026-10-24). A fill posted after the window is a release of its own (the judge's `repeated` - rule already knows reposts). At the block's end the releases the window - hasn't closed on are written **provisionally** (`settled_at` null) and - flagged `edge`, as are the finished releases posted within 24 h of the - block's first frontier; the block merger (replaces the fragment merger) - joins the edge releases of adjacent blocks of a group by identity key - once both are done, finishes them and clears the flag. The settler goes: - `settled_at` is the finish. The **tail** needs no path of its own: a - tail lease's open releases at its end are such provisional edges, the - next lease's continuation joins them. + rule already knows reposts). At a lease's end (the tail's poll, a worker + stopping; amended 2026-10-26, was: the block's end) the releases the + window hasn't closed on are written **provisionally** (`settled_at` + null, `finished: false`, flagged `edge`): a release being uploaded + shows in the catalog, and the carried window finishes the same row + later, since the release key is the identity, the group and the first + article; the store's finish path adds the files not sent before. No + merger: the carry does the joining. The settler goes: `settled_at` is + the finish. The **tail** needs no path of its own: its lease ends leave + such provisional releases, the next lease's continuation finishes them. 3. **Albums.** Files of one poster in one owner group whose subjects differ only by a number sequence, or share a directory-like prefix, posted @@ -445,7 +458,16 @@ releases, a third of them one poster's image dump (27.7 M rows, LOG.md a day; at a file per ~50 articles that is ~1 M open files of ~100 B in memory per block, and their segment lists (~25 B each) in the block's spool file, read back when a release is finished. Per worker thread a - few hundred MB at worst. + few hundred MB at worst. The carried window (amended 2026-10-26) has to + survive a restart (deploys restart the workers daily): its counters and + names are small (kilobytes for most groups, ~100 MB for the busiest), + the open files' segment lists are the bulk (~1.25 GB for a day of the + busiest group). A snapshot at every 2 M range would be ~600 B/article + 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. 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 @@ -603,11 +625,37 @@ indexer; CI runs `edge/bench-fixtures.sh` instead), the `incomplete` count window holds a day of a group's segments in memory (the busiest group's ~50 M); the spool file per block of the design is not built yet. -Next: albums in the edge (design 3), block leases and the planner (then -the store keys by the leased block and the lease columns leave `ranges`; -until then every range end leaves provisional edges), the block merger -(joins and finishes the edges), the prober writing `release_texts` kind 2, -the corpus replay harness. +### The replay harness (built 2026-10-26) + +`node ace replay:corpus --group [--dir CORPUS_DIR] [--from article] +[--limit ranges] [--batch-chunks 5] [--no-finish]` (commands/ +replay_corpus.ts; design 8): a group's complete corpus ranges in article +order through one window in the Rust edge (`edge-bench --group +--rules rules.json --wire -`: the window carried from range to range, the +end finished after the last range; `edge/src/lease.rs` has the lease JSON +the worker and the bench share) and the batch store into a scratch +catalog, as the worker would store them: each range leased to a node named +`replay`, its batches stored at its resume point (pulled from the bench's +stdout as the store keeps up), the range finished when its last batch is +in. It refuses a database not named `*_replay` or `*_scratch`. +The rules are the scratch catalog's current set and group priorities: +copy `groups` and `rule_sets` from the catalog for a faithful replay. The +report: releases finished, provisional, edges; complete, incomplete, +undecided; readable, with par2; files, segments as a share of the +headers; bytes per header of the NVMe tables (releases, release_texts, +posters, group_sets, rule counts and samples) and of those on +CHUNK_TABLESPACE (release_chunks, probe_candidates, release_inspections). +The app image carries `edge-bench` for it (Dockerfile); on homelab the +scratch catalog is a second database on bulk (`usenet_replay`, migrated +with the voyager-3 image, `groups` and `rule_sets` copied over), the +corpus mounted read-only: not run there yet (LOG.md 2026-10-26 (111) has +the first local numbers). + +Next: the replay on homelab over the 2016 window (the NVMe bytes per +header for real), albums in the edge (design 3), 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. Gone from the list: block +leases, the planner and the block merger (design 1, amended). ## Decisions diff --git a/commands/replay_corpus.ts b/commands/replay_corpus.ts new file mode 100644 index 0000000..b3492b5 --- /dev/null +++ b/commands/replay_corpus.ts @@ -0,0 +1,370 @@ +import { BaseCommand, flags } from '@adonisjs/core/ace' +import type { CommandOptions } from '@adonisjs/core/types/ace' +import { spawn } from 'node:child_process' +import { access, readdir, readFile, writeFile } from 'node:fs/promises' +import { tmpdir } from 'node:os' +import { join, resolve } from 'node:path' +import db from '@adonisjs/lucid/services/db' +import coordinatorConfig from '#config/coordinator' +import BatchStore from '#services/batch_store' +import Leases from '#services/leases' +import RuleSets from '#services/rule_sets' +import { fromWire, type WireBatch } from '#lib/edge/index' + +/** + * Only a database named like one of these is written: the replay leases and finishes ranges and fills the catalog. + * Not the test database: the rows left behind failed 27 functional tests (LOG.md 2026-10-26 (111)). + */ +const SCRATCH_DB = /_(replay|scratch)$/ +const NODE_NAME = 'replay' + +interface StoredRange { + id: number + start: number + end: number +} + +/** + * The replay harness (REBUILD.md, "Voyager 3", 8): a group's stored corpus ranges in article order through one + * window in the Rust edge (`edge-bench --group`, the window carried from range to range) and the batch store into a + * scratch catalog, as the worker would store them: each range leased to a node named "replay", its batches stored + * at its resume point, the range finished when its last batch is in. Then a report: releases finished, complete, + * partial and provisional, and the catalog's bytes per header (the NVMe tables apart from those on CHUNK_TABLESPACE). + * + * node ace replay:corpus --group alt.binaries.ath [--dir CORPUS_DIR] [--limit N] [--batch-chunks 5] + * + * Refuses a database not named *_replay or *_scratch. The rules are the catalog's current rule set + * (rule_sets, the groups' priorities): copy `groups` and `rule_sets` from the catalog for a faithful replay. + */ +export default class ReplayCorpus extends BaseCommand { + static commandName = 'replay:corpus' + static description = + "Replays a group's corpus ranges through the Rust edge and the batch store into a scratch catalog" + static options: CommandOptions = { startApp: true } + + @flags.string({ description: 'The group (its directory under the corpus)', required: true }) + declare group: string + + @flags.string({ description: 'The corpus directory (default: CORPUS_DIR)' }) + declare dir?: string + + @flags.number({ description: 'Replay this many ranges at most (default: all stored)' }) + declare limit?: number + + @flags.number({ description: 'Skip ranges ending before this article (continue a replay)' }) + declare from?: number + + @flags.number({ + description: 'Chunks of 100k articles per batch, as the worker posts them', + default: 5, + }) + declare batchChunks: number + + @flags.string({ + description: + 'The edge-bench binary (default: EDGE_BENCH, edge/target/release/edge-bench, /usr/local/bin/edge-bench)', + }) + declare bench?: string + + @flags.boolean({ + description: + 'Leave the window open at the end (the last ranges provisional), to continue later', + }) + declare noFinish: boolean + + async run() { + const database = db.connection().getReadClient().client.config.connection.database as string + if (!SCRATCH_DB.test(database)) { + this.logger.error( + `The replay writes the catalog: it only runs against a database named *_replay or *_scratch (this is "${database}")` + ) + this.exitCode = 1 + return + } + const dir = this.dir ?? coordinatorConfig.corpus.dir + if (!dir) { + this.logger.error('No corpus directory: --dir or CORPUS_DIR') + this.exitCode = 1 + return + } + const bench = await this.benchBinary() + if (!bench) { + this.logger.error( + 'No edge-bench binary found: build it (cargo build --release --manifest-path edge/Cargo.toml) or set EDGE_BENCH' + ) + this.exitCode = 1 + return + } + const groupDir = join(dir, this.group) + const stored = await this.storedRanges(groupDir) + if (stored.length === 0) { + this.logger.error(`${groupDir}: no complete range`) + this.exitCode = 1 + return + } + const nodeId = await this.nodeId() + const groupId = await this.groupId() + const ranges = await this.leaseRanges(groupId, nodeId, stored) + const rules = await RuleSets.current() + const rulesPath = join(tmpdir(), `replay-rules-${process.pid}.json`) + await writeFile(rulesPath, JSON.stringify(rules)) + this.logger.info( + `${this.group}: ${ranges.length} ranges (${ranges[0].start}–${ranges[ranges.length - 1].end}) through ${bench}, rules v${rules.version}, into ${database}.` + ) + + // edge-bench walks the group directory in article order itself; --from and --limit cut its list as ours. + const args = [ + '--group', + groupDir, + '--rules', + rulesPath, + '--batch-chunks', + String(this.batchChunks), + '--wire', + '-', + ] + if (this.from) args.push('--from', String(this.from)) + if (this.limit) args.push('--limit', String(this.limit)) + if (this.noFinish) args.push('--no-finish') + const started = Date.now() + const child = spawn(bench, args, { stdio: ['ignore', 'pipe', 'inherit'] }) + const totals = { + batches: 0, + headers: 0, + releases: 0, + newReleases: 0, + finalized: 0, + candidates: 0, + storeMs: 0, + } + let current: StoredRange | null = null + const storeLine = async (line: string) => { + const batch = fromWire(JSON.parse(line) as WireBatch) + if (!current || batch.from < current.start || batch.from > current.end) { + current = ranges.find((r) => batch.from >= r.start && batch.from <= r.end) ?? null + if (!current) throw new Error(`Batch at ${batch.from} is in no leased range`) + } + const at = Date.now() + const result = await BatchStore.store(current.id, nodeId, batch) + totals.storeMs += Date.now() - at + totals.batches++ + totals.headers += batch.headers + totals.releases += result.releases + totals.newReleases += result.newReleases + totals.finalized += result.finalized + totals.candidates += result.candidates + if (batch.next > current.end) { + await Leases.finish(current.id, nodeId) + const seconds = (Date.now() - started) / 1000 + this.logger.info( + ` range ${current.start}–${current.end} done: ${totals.headers.toLocaleString('en')} headers so far in ${seconds.toFixed(0)} s ` + + `(${Math.round(totals.headers / seconds).toLocaleString('en')}/s), ${totals.releases} releases stored (${totals.finalized} finished later), store ${(totals.storeMs / 1000).toFixed(0)} s` + ) + } + } + // Lines pulled as the store keeps up: the child's stdout only moves when a chunk is read, so batches don't pile up. + let rest = '' + let failure: Error | null = null + try { + for await (const chunk of child.stdout as AsyncIterable) { + rest += chunk.toString('utf8') + let nl: number + while ((nl = rest.indexOf('\n')) >= 0) { + const line = rest.slice(0, nl) + rest = rest.slice(nl + 1) + if (line.length > 0) await storeLine(line) + } + } + if (rest.trim().length > 0) await storeLine(rest) + } catch (error: any) { + failure = error + child.kill('SIGTERM') + } + const code = await new Promise((done) => + child.exitCode !== null ? done(child.exitCode) : child.on('exit', done) + ) + if (failure) { + this.logger.error(`Replay failed: ${failure.message}`) + this.exitCode = 1 + return + } + if (code !== 0) { + this.logger.error(`edge-bench exited with ${code}`) + this.exitCode = 1 + return + } + await this.report(ranges, Date.now() - started) + } + + private async benchBinary(): Promise { + const candidates = [ + this.bench, + process.env.EDGE_BENCH, + resolve('edge/target/release/edge-bench'), + '/usr/local/bin/edge-bench', + ].filter((p): p is string => !!p) + for (const path of candidates) { + if ( + await access(path).then( + () => true, + () => false + ) + ) + return path + } + return null + } + + /** The group directory's complete ranges (a range.json), in article order, cut by --from and --limit. */ + private async storedRanges(groupDir: string): Promise<{ start: number; end: number }[]> { + const names = await readdir(groupDir).catch(() => [] as string[]) + const ranges: { start: number; end: number }[] = [] + for (const name of names) { + const m = /^(\d+)-(\d+)$/.exec(name) + if (!m) continue + const json = await readFile(join(groupDir, name, 'range.json'), 'utf8').catch(() => null) + if (!json) continue + ranges.push({ start: Number(m[1]), end: Number(m[2]) }) + } + ranges.sort((a, b) => a.start - b.start) + let chosen = this.from ? ranges.filter((r) => r.end >= this.from!) : ranges + if (this.limit) chosen = chosen.slice(0, this.limit) + return chosen + } + + private async nodeId(): Promise { + const row = await db.from('nodes').where('name', NODE_NAME).first() + if (row) return Number(row.id) + const [inserted] = await db.table('nodes').insert({ name: NODE_NAME }).returning('id') + return Number(inserted.id) + } + + /** The group's row (inserted after the last priority when the scratch catalog doesn't have it). */ + private async groupId(): Promise { + const row = await db.from('groups').where('name', this.group).first() + if (row) return Number(row.id) + const { rows } = await db.rawQuery(`SELECT coalesce(max(priority), 0) + 1 AS next FROM groups`) + const [inserted] = await db + .table('groups') + .insert({ name: this.group, priority: Number(rows[0].next), active: true }) + .returning('id') + this.logger.info(`Group ${this.group} added with priority ${rows[0].next}.`) + return Number(inserted.id) + } + + /** The ranges as leased to the replay node, from their start (an earlier replay's rows are taken over). */ + private async leaseRanges( + groupId: number, + nodeId: number, + stored: { start: number; end: number }[] + ): Promise { + const out: StoredRange[] = [] + const now = new Date() + for (const r of stored) { + const lease = { + status: 'leased', + node_id: nodeId, + next_id: r.start, + headers: 0, + posted_low: null, + posted_high: null, + leased_at: now, + heartbeat_at: now, + done_at: null, + error: null, + capture: true, + } + const existing = await db + .from('ranges') + .where('group_id', groupId) + .where('start_id', r.start) + .first() + if (existing) { + await db + .from('ranges') + .where('id', existing.id) + .update({ ...lease, end_id: r.end, attempts: Number(existing.attempts) + 1 }) + out.push({ id: Number(existing.id), start: r.start, end: r.end }) + continue + } + const [inserted] = await db + .table('ranges') + .insert({ + ...lease, + group_id: groupId, + start_id: r.start, + end_id: r.end, + priority: 0, + attempts: 1, + }) + .returning('id') + out.push({ id: Number(inserted.id), start: r.start, end: r.end }) + } + return out + } + + /** What the replay left in the catalog, and its size per article. */ + private async report(ranges: StoredRange[], elapsedMs: number) { + const ids = ranges.map((r) => r.id) + const { rows: counts } = await db.rawQuery( + `SELECT count(*)::bigint AS releases, + count(settled_at)::bigint AS finished, + count(*) FILTER (WHERE settled_at IS NULL)::bigint AS provisional, + count(*) FILTER (WHERE edge)::bigint AS edges, + count(*) FILTER (WHERE complete)::bigint AS complete, + count(*) FILTER (WHERE complete = false)::bigint AS incomplete, + count(*) FILTER (WHERE complete IS NULL)::bigint AS undecided, + count(*) FILTER (WHERE readable)::bigint AS readable, + count(*) FILTER (WHERE has_par2)::bigint AS with_par2, + coalesce(sum(files), 0)::bigint AS files, + coalesce(sum(segments), 0)::bigint AS segments, + coalesce(sum(bytes), 0)::bigint AS bytes + FROM releases` + ) + const { rows: sources } = await db.rawQuery( + `SELECT source, count(*)::bigint AS n FROM releases GROUP BY source ORDER BY source` + ) + const { rows: headerRows } = await db.rawQuery( + `SELECT coalesce(sum(headers), 0)::bigint AS headers FROM ranges WHERE id = ANY(?::bigint[])`, + [ids] + ) + const headers = Number(headerRows[0].headers) + const { rows: sizes } = await db.rawQuery( + `SELECT CASE WHEN c.relname LIKE 'release_chunks%' THEN 'release_chunks' ELSE c.relname END AS name, + sum(pg_total_relation_size(c.oid))::bigint AS bytes + FROM pg_class c JOIN pg_namespace n ON n.oid = c.relnamespace + WHERE n.nspname = 'public' AND c.relkind = 'r' + AND (c.relname IN ('releases', 'release_texts', 'posters', 'group_sets', 'probe_candidates', 'rule_counts', 'rule_samples', 'blocks') + OR c.relname LIKE 'release_chunks%') + GROUP BY 1 ORDER BY 2 DESC` + ) + // On CHUNK_TABLESPACE (the HDD) in the catalog: the chunks (migration 3200), inspections and candidates (6000). + const hdd = new Set(['release_chunks', 'probe_candidates', 'release_inspections']) + let nvme = 0 + let other = 0 + const c = counts[0] + const fmt = (n: unknown) => Number(n).toLocaleString('en') + this.logger.info( + `Replay of ${ranges.length} ranges: ${fmt(headers)} headers in ${(elapsedMs / 1000).toFixed(0)} s (${fmt(Math.round(headers / Math.max(1, elapsedMs / 1000)))}/s).` + ) + this.logger.info( + `Releases ${fmt(c.releases)}: finished ${fmt(c.finished)}, provisional ${fmt(c.provisional)}, edges ${fmt(c.edges)}; ` + + `complete ${fmt(c.complete)}, incomplete ${fmt(c.incomplete)}, undecided ${fmt(c.undecided)}; readable ${fmt(c.readable)}, with par2 ${fmt(c.with_par2)}; ` + + `${fmt(c.files)} files, ${fmt(c.segments)} segments (${((Number(c.segments) / Math.max(1, headers)) * 100).toFixed(1)}% of the headers), ${(Number(c.bytes) / 1e12).toFixed(2)} TB.` + ) + this.logger.info( + `By source: ${sources.map((s: any) => `${s.source}: ${fmt(s.n)}`).join(', ')}.` + ) + for (const s of sizes) { + const bytes = Number(s.bytes) + if (hdd.has(s.name)) other += bytes + else nvme += bytes + this.logger.info( + ` ${s.name}: ${(bytes / 1e6).toFixed(1)} MB, ${(bytes / Math.max(1, headers)).toFixed(2)} B/header` + ) + } + this.logger.info( + `NVMe tables: ${(nvme / 1e6).toFixed(1)} MB, ${(nvme / Math.max(1, headers)).toFixed(2)} B/header; HDD tables (CHUNK_TABLESPACE): ${(other / 1e6).toFixed(1)} MB, ${(other / Math.max(1, headers)).toFixed(2)} B/header.` + ) + } +} diff --git a/deploy/compose.yaml b/deploy/compose.yaml index 50aa40f..bc0dc12 100644 --- a/deploy/compose.yaml +++ b/deploy/compose.yaml @@ -190,6 +190,21 @@ services: node-api: condition: service_started + # The replay harness into the scratch catalog usenet_replay on bulk (REBUILD.md, "The replay harness"): + # docker compose run --rm --no-deps replay --group alt.binaries.ath + replay: + <<: *app + profiles: [replay] + restart: 'no' + entrypoint: ['node', 'ace', 'replay:corpus'] + command: ['--help'] + environment: + <<: *app-env + DB_DATABASE: usenet_replay + volumes: + - ./corpus:/corpus:ro + depends_on: [] + # One-off ace commands: docker compose run --rm ace releases:duplicates ace: <<: *app diff --git a/edge/README.md b/edge/README.md index 303e7b5..274aa88 100644 --- a/edge/README.md +++ b/edge/README.md @@ -113,11 +113,29 @@ files, segments, payload size, candidates, fragments and samples. The summary can hide compensating differences, so compare the outputs themselves. -## Comparing the output +## Replaying a group through one window + +A corpus group directory (`//-/`, each range with +its `range.json`) is replayed in article order with the window carried from +range to range, the end finished after the last range (REBUILD.md, "The +replay harness"): + +```sh +./target/release/edge-bench --group /corpus/alt.binaries.ath --rules rules.json --batch-chunks 5 --wire - > batches.jsonl +# --from
skips ranges ending before it, --limit stops after n ranges, --no-finish leaves the window open +``` + +`--rules` takes the lease's rules JSON (`src/lease.rs`: the catalog's rule +set and group priorities, as `replay:corpus` writes it); without it the +defaults with only the group indexed apply. `--wire -` writes each batch as +the worker posts it to stdout, one JSON per line, and the report to stderr; +`node ace replay:corpus` reads them into the batch store. Each range end +prints the frontier, the releases out and the releases and files still open. + +## The dump ```sh ./target/release/edge-bench /path/to/capture --dump /tmp/out.jsonl -cd .. && node --import=@poppinss/ts-exec edge/check.ts /path/to/capture /tmp/out.jsonl ``` `--dump` writes every output of every batch as JSON lines: diff --git a/edge/src/bin/edge-node.rs b/edge/src/bin/edge-node.rs index 0e2f134..1ba9bd1 100644 --- a/edge/src/bin/edge-node.rs +++ b/edge/src/bin/edge-node.rs @@ -9,7 +9,7 @@ //! (the next worker resumes at the last stored batch). use edge::indexer::{Batch, RangeIndexer}; -use edge::names::Rules; +use edge::lease::LeaseJson; use edge::wire; use flate2::write::GzEncoder; use serde::Deserialize; @@ -34,63 +34,6 @@ fn log(level: &str, message: &str) { println!("[ {level} ] {message}"); } -#[derive(Deserialize)] -struct LeaseJson { - range: RangeJson, - rules: RulesJson, -} - -#[derive(Deserialize)] -struct RangeJson { - id: i64, - group: String, - start: i64, - end: i64, - next: i64, -} - -#[derive(Deserialize)] -#[serde(rename_all = "camelCase")] -struct RulesJson { - version: i64, - priorities: HashMap, - #[serde(default)] - dropped_signatures: Vec, - #[serde(default)] - dropped_tools: Option>, - #[serde(default)] - per_article_max_parts: Option, - #[serde(default)] - sample_percent: Option, - #[serde(default)] - drop_executables: Option, - #[serde(default)] - dropped_posters: Option>, -} - -impl RulesJson { - fn rules(self) -> Rules { - let mut rules = Rules::defaults(self.priorities.into_iter().collect()); - rules.version = self.version; - rules.dropped_signatures = self.dropped_signatures.into_iter().collect(); - if let Some(tools) = self.dropped_tools { - rules.dropped_tools = tools.into_iter().collect(); - } - if let Some(n) = self.per_article_max_parts { - rules.per_article_max_parts = n; - } - if let Some(p) = self.sample_percent { - rules.sample_percent = p; - } - if let Some(e) = self.drop_executables { - rules.drop_executables = e; - } - if let Some(posters) = self.dropped_posters { - rules.dropped_posters = posters; - } - rules - } -} #[derive(Deserialize)] struct StoredJson { diff --git a/edge/src/indexer.rs b/edge/src/indexer.rs index 50a9091..2559602 100644 --- a/edge/src/indexer.rs +++ b/edge/src/indexer.rs @@ -832,6 +832,16 @@ impl RangeIndexer { self.frontier.value() } + /// Releases the window still holds (edge-bench's progress line). + pub fn open_releases(&self) -> usize { + self.open.values().map(|u| u.len()).sum() + } + + /// Files still short of parts, or complete and not yet in a release. + pub fn open_files(&self) -> usize { + self.aggregator.open.values().map(|f| f.len()).sum() + } + pub fn take(&mut self) -> Batch { self.scanner.files.clear(); let next = self.batch.next; diff --git a/edge/src/lease.rs b/edge/src/lease.rs new file mode 100644 index 0000000..bc4a24e --- /dev/null +++ b/edge/src/lease.rs @@ -0,0 +1,65 @@ +//! A lease as the coordinator sends it (lib/edge/lease.ts): the range and the rules, JSON with the TypeScript +//! names. The edge-node worker takes one per range; edge-bench's replay takes the rules alone (`--rules`). + +use crate::names::Rules; +use serde::Deserialize; +use std::collections::HashMap; + +#[derive(Deserialize)] +pub struct LeaseJson { + pub range: RangeJson, + pub rules: RulesJson, +} + +#[derive(Deserialize)] +pub struct RangeJson { + pub id: i64, + pub group: String, + pub start: i64, + pub end: i64, + pub next: i64, +} + +#[derive(Deserialize)] +#[serde(rename_all = "camelCase")] +pub struct RulesJson { + pub version: i64, + pub priorities: HashMap, + #[serde(default)] + pub dropped_signatures: Vec, + #[serde(default)] + pub dropped_tools: Option>, + #[serde(default)] + pub per_article_max_parts: Option, + #[serde(default)] + pub sample_percent: Option, + #[serde(default)] + pub drop_executables: Option, + #[serde(default)] + pub dropped_posters: Option>, +} + +impl RulesJson { + /// The rules with today's defaults under them (a field the stored set doesn't have keeps the default). + pub fn rules(self) -> Rules { + let mut rules = Rules::defaults(self.priorities.into_iter().collect()); + rules.version = self.version; + rules.dropped_signatures = self.dropped_signatures.into_iter().collect(); + if let Some(tools) = self.dropped_tools { + rules.dropped_tools = tools.into_iter().collect(); + } + if let Some(n) = self.per_article_max_parts { + rules.per_article_max_parts = n; + } + if let Some(p) = self.sample_percent { + rules.sample_percent = p; + } + if let Some(e) = self.drop_executables { + rules.drop_executables = e; + } + if let Some(posters) = self.dropped_posters { + rules.dropped_posters = posters; + } + rules + } +} diff --git a/edge/src/lib.rs b/edge/src/lib.rs index 540a70a..0302556 100644 --- a/edge/src/lib.rs +++ b/edge/src/lib.rs @@ -5,5 +5,6 @@ pub mod codec; pub mod indexer; pub mod js; +pub mod lease; pub mod names; pub mod wire; diff --git a/edge/src/main.rs b/edge/src/main.rs index b025212..ab1ee4f 100644 --- a/edge/src/main.rs +++ b/edge/src/main.rs @@ -1,10 +1,11 @@ -//! The edge worker's indexing in Rust, on a range captured by `edge:bench -//! --capture` (REBUILD.md, part 2: TypeScript or Rust). Same work as -//! edge:bench's replay: inflate each chunk, scan it, aggregate files, cut off -//! idle ones, and every N chunks judge and encode the releases. `--dump` -//! writes every output as JSON lines for edge/check.ts. +//! The Rust indexer over stored chunks (REBUILD.md, part 2 and "Voyager 3"): a range captured by `edge:bench +//! --capture` or stored in the corpus, or a group's corpus ranges in order through one window (`--group`, the +//! replay harness behind `replay:corpus`). Same work as the edge-node worker: inflate each chunk, scan it, +//! aggregate files, finish releases through the window, and every N chunks judge and encode them. `--dump` +//! writes every output as JSON lines; `--wire` the batches as the worker posts them (`-`: stdout). //! -//! edge-bench [--batch-chunks 10] [--idle 5] [--dump out.jsonl] [--wire batches.jsonl] +//! edge-bench | --group [--from article] [--limit ranges] [--rules rules.json] +//! [--batch-chunks 10] [--idle 5] [--no-finish] [--dump out.jsonl] [--wire batches.jsonl|-] [--releases out.jsonl] use base64::Engine; use edge::indexer::{self, Batch, RangeIndexer}; @@ -167,99 +168,210 @@ fn dump_batch(batch: &Batch, out: &mut impl Write) { } } +/// A stored range: its directory and article span (range.json). +struct Range { + dir: String, + start: i64, + end: i64, +} + +/// The range in a directory with a range.json, and its group. +fn range_in(dir: &str) -> Option<(String, Range)> { + let json = std::fs::read_to_string(format!("{dir}/range.json")).ok()?; + Some((json_field(&json, "group"), Range { dir: dir.to_string(), start: json_field(&json, "start").parse().ok()?, end: json_field(&json, "end").parse().ok()? })) +} + +/// A group directory's complete ranges in article order (the corpus layout: `/-/`). +fn ranges_in(group_dir: &str) -> (String, Vec) { + let mut group = String::new(); + let mut ranges: Vec = Vec::new(); + for entry in std::fs::read_dir(group_dir).unwrap_or_else(|e| panic!("{group_dir}: {e}")) { + let path = entry.unwrap().path(); + if !path.is_dir() { + continue; + } + if let Some((g, range)) = range_in(path.to_str().unwrap()) { + group = g; + ranges.push(range); + } + } + ranges.sort_by_key(|r| r.start); + (group, ranges) +} + +fn date(ms: Option) -> String { + match ms.and_then(chrono::DateTime::from_timestamp_millis) { + Some(t) => t.format("%Y-%m-%d %H:%M").to_string(), + None => "-".to_string(), + } +} + +/// Files waiting in finished releases after which a batch goes out whatever its chunks (edge-node's limit). +const MAX_PENDING_FILES: usize = 100_000; + fn main() { let args: Vec = std::env::args().collect(); - let dir = args.get(1).expect("usage: edge-bench [--batch-chunks N] [--idle N] [--dump file]").clone(); + let usage = "usage: edge-bench | --group [--from article] [--limit ranges] [--rules rules.json] [--batch-chunks N] [--idle N] [--no-finish] [--dump file] [--wire file|-] [--releases file]"; let flag = |name: &str| args.iter().position(|a| a == name).and_then(|i| args.get(i + 1)).cloned(); + let has = |name: &str| args.iter().any(|a| a == name); let batch_chunks: usize = flag("--batch-chunks").map(|v| v.parse().unwrap()).unwrap_or(10); let idle: u64 = flag("--idle").map(|v| v.parse().unwrap()).unwrap_or(5); let mut dump = flag("--dump").map(|p| std::io::BufWriter::new(std::fs::File::create(p).unwrap())); - // Each batch as the worker posts it (wire.rs), one JSON per line: check.ts compares it with toWire. - let mut wire_out = flag("--wire").map(|p| std::io::BufWriter::new(std::fs::File::create(p).unwrap())); + // Each batch as the worker posts it (wire.rs), one JSON per line: a file, or stdout (`-`) for replay:corpus. + let wire_path = flag("--wire"); + let wire_stdout = wire_path.as_deref() == Some("-"); + let mut wire_out: Option> = match wire_path.as_deref() { + Some("-") => Some(Box::new(std::io::BufWriter::with_capacity(1 << 20, std::io::stdout().lock()))), + Some(p) => Some(Box::new(std::io::BufWriter::new(std::fs::File::create(p).unwrap()))), + None => None, + }; // One line per release chunk without its payload, with its first subjects (deploy/corpus_sweep.py). let mut releases_out = flag("--releases").map(|p| std::io::BufWriter::new(std::fs::File::create(p).unwrap())); - let range = std::fs::read_to_string(format!("{dir}/range.json")).unwrap(); - let group = json_field(&range, "group"); - let start: i64 = json_field(&range, "start").parse().unwrap(); - let end: i64 = json_field(&range, "end").parse().unwrap(); - let mut files: Vec = std::fs::read_dir(&dir).unwrap().map(|e| e.unwrap().file_name().into_string().unwrap()).filter(|f| f.ends_with(".gz")).collect(); - files.sort(); + // One range, or a group's ranges in order through one window (the window carries over from range to range; + // the end is only finished after the last one, unless --no-finish: a replay to be continued). + let (group, ranges) = match flag("--group") { + Some(dir) => { + let (group, mut ranges) = ranges_in(&dir); + // --from: ranges ending before it are skipped (a replay continued); --limit: this many at most. + if let Some(from) = flag("--from").map(|v| v.parse::().unwrap()) { + ranges.retain(|r| r.end >= from); + } + if let Some(limit) = flag("--limit").map(|v| v.parse::().unwrap()) { + ranges.truncate(limit); + } + (group, ranges) + } + None => { + let positional = (1..args.len()).find(|&i| !args[i].starts_with("--") && !args[i - 1].starts_with("--")).map(|i| args[i].clone()); + let dir = positional.expect(usage); + let (group, range) = range_in(&dir).unwrap_or_else(|| panic!("{dir}: no range.json")); + (group, vec![range]) + } + }; + if ranges.is_empty() { + eprintln!("no complete range found"); + std::process::exit(2); + } + let finish = !has("--no-finish"); + let rules = match flag("--rules") { + Some(path) => { + let json = std::fs::read_to_string(&path).unwrap_or_else(|e| panic!("{path}: {e}")); + serde_json::from_str::(&json).unwrap_or_else(|e| panic!("{path}: {e}")).rules() + } + None => { + let mut priorities = FxHashMap::default(); + priorities.insert(group.clone(), 1); + names::Rules::defaults(priorities) + } + }; + // The report goes to stderr when the wire takes stdout. + let say = |line: String| { + if wire_stdout { + eprintln!("{line}"); + } else { + println!("{line}"); + } + }; - let mut priorities = FxHashMap::default(); - priorities.insert(group.clone(), 1); - let mut indexer = RangeIndexer::new(&group, start, end, start, names::Rules::defaults(priorities), idle); + let (start, end) = (ranges[0].start, ranges[ranges.len() - 1].end); + let mut indexer = RangeIndexer::new(&group, start, end, start, rules, idle); let (mut inflate_s, mut index_s) = (0.0, 0.0); - let (mut compressed, mut text_bytes) = (0usize, 0usize); + let (mut compressed, mut text_bytes, mut chunks) = (0usize, 0usize, 0usize); let mut out = (0u64, 0usize, 0usize, 0usize, 0usize, 0usize, 0usize); // headers, releases, files, segments, payload, candidates, samples + let mut finished = 0usize; let mut keys = std::collections::HashSet::new(); let mut decompressor = libdeflater::Decompressor::new(); let wall = Instant::now(); - for (i, file) in files.iter().enumerate() { - let raw = std::fs::read(format!("{dir}/{file}")).unwrap(); - compressed += raw.len(); - let t = Instant::now(); - let data = inflate(&raw, &mut decompressor); - inflate_s += t.elapsed().as_secs_f64(); - text_bytes += data.len(); - let t = Instant::now(); - let text = String::from_utf8_lossy(&data); - let next: i64 = file.split('-').nth(2).unwrap().trim_end_matches(".gz").parse::().unwrap() + 1; - indexer.add_text(&text, next); - let last = i + 1 == files.len(); - if last { - indexer.finish(); - } - if !last && (i + 1) % batch_chunks != 0 { + let mut since_batch = 0usize; + for (ri, range) in ranges.iter().enumerate() { + let range_wall = Instant::now(); + let headers_before = out.0; + let mut files: Vec = std::fs::read_dir(&range.dir).unwrap().map(|e| e.unwrap().file_name().into_string().unwrap()).filter(|f| f.ends_with(".gz")).collect(); + files.sort(); + for (i, file) in files.iter().enumerate() { + let raw = std::fs::read(format!("{}/{file}", range.dir)).unwrap(); + compressed += raw.len(); + let t = Instant::now(); + let data = inflate(&raw, &mut decompressor); + inflate_s += t.elapsed().as_secs_f64(); + text_bytes += data.len(); + chunks += 1; + let t = Instant::now(); + let text = String::from_utf8_lossy(&data); + let next: i64 = file.split('-').nth(2).unwrap().trim_end_matches(".gz").parse::().unwrap() + 1; + indexer.add_text(&text, next); + since_batch += 1; + let range_end = i + 1 == files.len(); + let last = range_end && ri + 1 == ranges.len(); + if last && finish { + indexer.finish(); + } + if !range_end && since_batch < batch_chunks && indexer.pending() < MAX_PENDING_FILES { + index_s += t.elapsed().as_secs_f64(); + continue; + } + let batch = indexer.take(); + since_batch = 0; index_s += t.elapsed().as_secs_f64(); - continue; + out.0 += batch.headers; + out.1 += batch.releases.len(); + for r in &batch.releases { + keys.insert(r.key.clone()); + out.2 += r.files; + out.3 += r.segments; + out.4 += r.payload.len(); + finished += r.finished as usize; + } + out.5 += batch.candidates.len(); + out.6 += batch.samples.len(); + if let Some(d) = dump.as_mut() { + dump_batch(&batch, d); + } + if let Some(w) = wire_out.as_mut() { + w.write_all(&wire::to_json(&batch)).unwrap(); + w.write_all(b"\n").unwrap(); + w.flush().unwrap(); + } + if let Some(w) = releases_out.as_mut() { + dump_releases(&batch, &group, w); + } } - let batch = indexer.take(); - index_s += t.elapsed().as_secs_f64(); - out.0 += batch.headers; - out.1 += batch.releases.len(); - for r in &batch.releases { - keys.insert(r.key.clone()); - out.2 += r.files; - out.3 += r.segments; - out.4 += r.payload.len(); - } - out.5 += batch.candidates.len(); - out.6 += batch.samples.len(); - if let Some(d) = dump.as_mut() { - dump_batch(&batch, d); - } - if let Some(w) = wire_out.as_mut() { - w.write_all(&wire::to_json(&batch)).unwrap(); - w.write_all(b"\n").unwrap(); - } - if let Some(w) = releases_out.as_mut() { - dump_releases(&batch, &group, w); + if ranges.len() > 1 { + eprintln!( + "{} {}–{}: {} headers in {:.1} s; frontier {}; {} releases out ({} finished), {} releases and {} files open; RSS {} MB", + group, + range.start, + range.end, + out.0 - headers_before, + range_wall.elapsed().as_secs_f64(), + date(indexer.frontier()), + out.1, + finished, + indexer.open_releases(), + indexer.open_files(), + peak_rss_mb() + ); } } let total = inflate_s + index_s; #[allow(static_mut_refs)] let p = unsafe { indexer::PROF }; eprintln!(" of which aggregate {:.2} s, evict+cutoff {:.2} s, judge+encode {:.2} s", p[3] as f64 / 1e9, p[4] as f64 / 1e9, p[5] as f64 / 1e9); - println!("{group} {start}–{end}: {} chunks, {} MB compressed, {} MB text", files.len(), compressed / 1_000_000, text_bytes / 1_000_000); - println!( - "{} headers in {:.2} s ({:.0}/s on one core; wall {:.2} s)", - out.0, - total, - out.0 as f64 / total, - wall.elapsed().as_secs_f64() - ); - println!(" inflate {inflate_s:.2} s, index {index_s:.2} s; peak RSS {} MB", peak_rss_mb()); - println!( - " kept: {} release chunks ({} releases), {} files, {} segments, payload {:.1} MB ({:.1} B/segment); {} candidates, {} samples", + say(format!("{group} {start}–{end}: {} ranges, {chunks} chunks, {} MB compressed, {} MB text", ranges.len(), compressed / 1_000_000, text_bytes / 1_000_000)); + say(format!("{} headers in {:.2} s ({:.0}/s on one core; wall {:.2} s)", out.0, total, out.0 as f64 / total, wall.elapsed().as_secs_f64())); + say(format!(" inflate {inflate_s:.2} s, index {index_s:.2} s; peak RSS {} MB", peak_rss_mb())); + say(format!( + " kept: {} release chunks ({} releases, {} finished), {} files, {} segments, payload {:.1} MB ({:.1} B/segment); {} candidates, {} samples", out.1, keys.len(), + finished, out.2, out.3, out.4 as f64 / 1e6, out.4 as f64 / out.3.max(1) as f64, out.5, out.6 - ); + )); }