diff --git a/REBUILD.md b/REBUILD.md index 2c3c4cf..9897a75 100644 --- a/REBUILD.md +++ b/REBUILD.md @@ -628,7 +628,7 @@ window holds a day of a group's segments in memory (the busiest group's ### 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/ +[--to 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 @@ -646,10 +646,16 @@ 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). +scratch catalog is a second database (`usenet_replay`, default tablespace +`nvme` like the catalog, the chunk tables on `pg_default`; migrated with +the voyager-3 image built from the deploy cache's repo, `groups` and +`rule_sets` copied over), the corpus mounted read-only. +`deploy/replay-window.sh` runs a window there: each group whose window +ranges are all stored is replayed once with the window's bounds as +`--from`/`--to` (the corpus holds other windows of the same group), 4 +groups at a time, logs and `.done` markers in `.cache/replay/`, looping +until the window is done (LOG.md 2026-10-26 (111) has the first local +numbers, (112) the homelab run). 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 diff --git a/commands/replay_corpus.ts b/commands/replay_corpus.ts index b3492b5..0868291 100644 --- a/commands/replay_corpus.ts +++ b/commands/replay_corpus.ts @@ -31,7 +31,7 @@ interface StoredRange { * 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] + * node ace replay:corpus --group alt.binaries.ath [--dir CORPUS_DIR] [--from A] [--to B] [--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. @@ -54,6 +54,11 @@ export default class ReplayCorpus extends BaseCommand { @flags.number({ description: 'Skip ranges ending before this article (continue a replay)' }) declare from?: number + @flags.number({ + description: "Skip ranges starting after this article (one window of the group's history)", + }) + declare to?: number + @flags.number({ description: 'Chunks of 100k articles per batch, as the worker posts them', default: 5, @@ -112,7 +117,7 @@ export default class ReplayCorpus extends BaseCommand { `${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. + // edge-bench walks the group directory in article order itself; --from, --to and --limit cut its list as ours. const args = [ '--group', groupDir, @@ -124,6 +129,7 @@ export default class ReplayCorpus extends BaseCommand { '-', ] if (this.from) args.push('--from', String(this.from)) + if (this.to) args.push('--to', String(this.to)) if (this.limit) args.push('--limit', String(this.limit)) if (this.noFinish) args.push('--no-finish') const started = Date.now() @@ -215,7 +221,7 @@ export default class ReplayCorpus extends BaseCommand { return null } - /** The group directory's complete ranges (a range.json), in article order, cut by --from and --limit. */ + /** The group directory's complete ranges (a range.json), in article order, cut by --from, --to 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 }[] = [] @@ -228,6 +234,7 @@ export default class ReplayCorpus extends BaseCommand { } ranges.sort((a, b) => a.start - b.start) let chosen = this.from ? ranges.filter((r) => r.end >= this.from!) : ranges + if (this.to) chosen = chosen.filter((r) => r.start <= this.to!) if (this.limit) chosen = chosen.slice(0, this.limit) return chosen } diff --git a/deploy/replay-window.sh b/deploy/replay-window.sh new file mode 100755 index 0000000..68fb36c --- /dev/null +++ b/deploy/replay-window.sh @@ -0,0 +1,69 @@ +#!/bin/bash +# A corpus window replayed on homelab (REBUILD.md, "The replay harness"): every group whose window ranges (the +# ranges marked `capture` at PRIORITY in the catalog, 0 for 2016) are all stored in the corpus is replayed once +# into the scratch catalog usenet_replay through the voyager-3 app image, PARALLEL groups at a time, the +# window's bounds passed as --from/--to (the corpus holds other windows of the same group). The group's log +# ends in .cache/replay/.log, a .done marker next to it (.failed when the replay failed: remove it to +# retry). Loops until every group of the window is done (the corpus fill runs meanwhile), so run it detached +# from the deployment root: +# setsid nohup deploy/replay-window.sh > .cache/replay/driver.log 2>&1 & +# No new group starts while the nightly pause is on (.cache/paused); the running ones finish. +set -u +ROOT=${ROOT:-/mnt/bulk/usenet-rebuild} +PRIORITY=${PRIORITY:-0} +PARALLEL=${PARALLEL:-4} +MEMORY=${MEMORY:-12g} +IMAGE=${IMAGE:-usenet-voyager3:latest} +OUT=$ROOT/.cache/replay +mkdir -p "$OUT" +cd "$ROOT" || exit 1 + +say() { echo "$(date -Is) $*"; } +catalog() { docker compose exec -T postgres psql -U rebuild -d usenet_rebuild -X -q -At -F ' ' -c "$1"; } + +# The window's groups: name, first article, last article, whether every range is stored; smallest first. +window() { + catalog "SELECT g.name, min(r.start_id), max(r.end_id), (count(*) = count(r.stored_at))::int + FROM ranges r JOIN groups g ON g.id = r.group_id + WHERE r.capture AND r.priority = $PRIORITY + GROUP BY g.name ORDER BY sum(r.end_id - r.start_id + 1)" +} + +replay() { + local group=$1 from=$2 to=$3 + say "$group: replaying $from–$to" + if docker run --rm --name "replay-${group//./-}" --network usenet-rebuild --env-file .env \ + -e NODE_ENV=production -e DB_HOST=postgres -e DB_PORT=5432 -e DB_DATABASE=usenet_replay -e CORPUS_DIR=/corpus \ + -v "$ROOT/corpus:/corpus:ro" --memory "$MEMORY" "$IMAGE" \ + node ace replay:corpus --group "$group" --from "$from" --to "$to" > "$OUT/$group.log" 2>&1; then + touch "$OUT/$group.done" + say "$group: done ($(grep -o 'Releases [0-9,]*' "$OUT/$group.log" | tail -1))" + else + touch "$OUT/$group.failed" + say "$group: FAILED (see $OUT/$group.log)" + fi + rm -f "$OUT/$group.running" +} + +while true; do + todo=0 + while read -r group from to complete; do + [ -n "$group" ] || continue + [ -e "$OUT/$group.done" ] && continue + [ -e "$OUT/$group.failed" ] && continue + todo=$((todo + 1)) + [ "$complete" = 1 ] || continue + [ -e "$OUT/$group.running" ] && continue + [ -e "$ROOT/.cache/paused" ] && continue + while [ "$(find "$OUT" -name '*.running' | wc -l)" -ge "$PARALLEL" ]; do sleep 30; done + touch "$OUT/$group.running" + replay "$group" "$from" "$to" & + sleep 5 + done <<< "$(window)" + if [ "$todo" = 0 ]; then + wait + say "every group of the window at priority $PRIORITY is replayed" + exit 0 + fi + sleep 300 +done diff --git a/edge/README.md b/edge/README.md index 274aa88..9245ddd 100644 --- a/edge/README.md +++ b/edge/README.md @@ -122,7 +122,8 @@ 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 +# --from
skips ranges ending before it, --to
ranges starting after it (one window of the +# group's history), --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 diff --git a/edge/src/main.rs b/edge/src/main.rs index ab1ee4f..80df678 100644 --- a/edge/src/main.rs +++ b/edge/src/main.rs @@ -4,7 +4,7 @@ //! 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 | --group [--from article] [--limit ranges] [--rules rules.json] +//! edge-bench | --group [--from article] [--to 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; @@ -211,7 +211,7 @@ const MAX_PENDING_FILES: usize = 100_000; fn main() { let args: Vec = std::env::args().collect(); - 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 usage = "usage: edge-bench | --group [--from article] [--to 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); @@ -233,10 +233,14 @@ fn main() { 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. + // --from: ranges ending before it are skipped (a replay continued); --to: ranges starting after it (one + // window of the group's history, the corpus holding others); --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(to) = flag("--to").map(|v| v.parse::().unwrap()) { + ranges.retain(|r| r.start <= to); + } if let Some(limit) = flag("--limit").map(|v| v.parse::().unwrap()) { ranges.truncate(limit); }