diff --git a/Dockerfile b/Dockerfile index e3e5859..a2490d8 100644 --- a/Dockerfile +++ b/Dockerfile @@ -32,10 +32,12 @@ WORKDIR /src COPY edge ./ RUN --mount=type=cache,target=/usr/local/cargo/registry \ --mount=type=cache,target=/src/target \ - cargo build --release --locked --bin edge-node \ - && cp target/release/edge-node /usr/local/bin/edge-node + cargo build --release --locked --bin edge-node --bin edge-bench \ + && cp target/release/edge-node target/release/edge-bench /usr/local/bin/ FROM debian:bookworm-slim AS edge COPY --from=edge-build /usr/local/bin/edge-node /usr/local/bin/edge-node +# edge-bench replays the corpus for the grouping sweeps (deploy/corpus_sweep.py). +COPY --from=edge-build /usr/local/bin/edge-bench /usr/local/bin/edge-bench USER 1000:1000 ENTRYPOINT ["/usr/local/bin/edge-node"] diff --git a/REBUILD.md b/REBUILD.md index 0bf6c1f..32915fc 100644 --- a/REBUILD.md +++ b/REBUILD.md @@ -208,6 +208,19 @@ of other groups whose counts show crossposts skipped for it (part 5). Start: the 1000 largest `alt.binaries.*` groups by LIST ACTIVE (~266 k ranges of 2 M articles; boneless alone 80 k). +## Corpus + +The grouping sweeps' dataset (2026-10-12; LOG.md 2026-10-12 (12)): the XOVER +chunks of ranges marked `capture`, kept on the HDD in +`corpus//-/` (edge-bench's layout, `range.json` once +complete), up to 300 GB. The prefetcher stores them as it fetches and reads +them from there afterwards, so a reset re-indexes them without the news +server. `deploy/corpus_sweep.py` marks more ranges each sweep (`capture`, +spread over groups and eras, the top-100 groups first), replays every stored +range with the deployed build's edge-bench (`replay`, ~seconds a range) and +reports suspects and the change since the previous build (`report`). A rule +change is judged on the whole corpus before a reset. + ## Phase 2 Built: the prober (2026-10-11; `probe:run`, LOG.md 2026-10-11 (2)). It diff --git a/app/services/corpus.ts b/app/services/corpus.ts new file mode 100644 index 0000000..a5577de --- /dev/null +++ b/app/services/corpus.ts @@ -0,0 +1,72 @@ +import { mkdir, readFile, readdir, rename, rm, writeFile } from 'node:fs/promises' +import { join } from 'node:path' +import { unzip } from 'node:zlib' +import { promisify } from 'node:util' +import coordinatorConfig from '#config/coordinator' +import type { ChunkSpan } from '#services/spool' + +const inflate = promisify(unzip) + +export interface CorpusRange { + group: string + start: number + end: number +} + +/** + * Stored XOVER chunks of the ranges marked `capture` (REBUILD.md, corpus): + * `//-/--.gz` with a range.json once + * every chunk is there, the layout edge-bench reads (edge/fixtures). The + * grouping sweeps replay them with each build's rules instead of waiting for + * a re-index; the prefetcher reads a stored chunk instead of fetching it + * again. Off without CORPUS_DIR. + */ +class Corpus { + get dir(): string | undefined { + return coordinatorConfig.corpus.dir + } + + private rangeDir(range: CorpusRange) { + return join(this.dir!, range.group, `${range.start}-${range.end}`) + } + + private path(range: CorpusRange, span: ChunkSpan) { + const index = Math.floor((span.from - range.start) / coordinatorConfig.chunkArticles) + return join(this.rangeDir(range), `${String(index).padStart(4, '0')}-${span.from}-${span.to}.gz`) + } + + /** A stored chunk, or null. */ + async read(range: CorpusRange, span: ChunkSpan): Promise { + if (!this.dir) return null + return readFile(this.path(range, span)).catch(() => null) + } + + /** + * Stores a chunk that inflates (a reply cut short isn't kept), and the + * range.json once the range is complete (`total` chunks). + */ + async write(range: CorpusRange, span: ChunkSpan, data: Buffer, total: number) { + if (!this.dir) return + try { + await inflate(data) + } catch { + return + } + const dir = this.rangeDir(range) + const file = this.path(range, span) + await mkdir(dir, { recursive: true }) + await writeFile(`${file}.part`, data) + await rename(`${file}.part`, file) + const chunks = (await readdir(dir)).filter((f) => f.endsWith('.gz')).length + if (chunks >= total) await writeFile(join(dir, 'range.json'), JSON.stringify(range)) + } + + /** A chunk a worker couldn't inflate after all: not served again, and the range not complete. */ + async removeChunk(range: CorpusRange, span: ChunkSpan) { + if (!this.dir) return + await rm(this.path(range, span), { force: true }) + await rm(join(this.rangeDir(range), 'range.json'), { force: true }) + } +} + +export default new Corpus() diff --git a/app/services/leases.ts b/app/services/leases.ts index a10b555..f6fa082 100644 --- a/app/services/leases.ts +++ b/app/services/leases.ts @@ -2,6 +2,7 @@ import db from '@adonisjs/lucid/services/db' import coordinatorConfig from '#config/coordinator' import RuleSets from '#services/rule_sets' import Spool from '#services/spool' +import Corpus from '#services/corpus' import type { Lease } from '#lib/edge/index' export class LeaseLost extends Error {} @@ -66,7 +67,13 @@ class Leases { async corrupt(rangeId: number, nodeId: number, from: number) { const range = await this.leased(rangeId, nodeId) const span = Spool.chunksOf({ start: Number(range.start_id), end: Number(range.end_id) }).find((s) => s.from === from) - if (span) await Spool.removeChunk(rangeId, span) + if (span) { + await Spool.removeChunk(rangeId, span) + if (range.capture) { + const group = await db.from('groups').where('id', range.group_id).select('name').first() + await Corpus.removeChunk({ group: group.name, start: Number(range.start_id), end: Number(range.end_id) }, span) + } + } const plain: number[] = (range.plain_chunks ?? []).map(Number) // Reported once already: the compressed copy may have been in flight (or the prefetcher's targets not // refreshed yet) and spooled again (range 161820, LOG.md 2026-10-11), so a second try plain first. diff --git a/app/services/prefetcher.ts b/app/services/prefetcher.ts index dedb136..f7892b1 100644 --- a/app/services/prefetcher.ts +++ b/app/services/prefetcher.ts @@ -3,6 +3,7 @@ import logger from '@adonisjs/core/services/logger' import coordinatorConfig from '#config/coordinator' import NntpPool from '#services/nntp_pool' import Spool, { type ChunkSpan } from '#services/spool' +import Corpus from '#services/corpus' import { hedge, type Request, type XoverClient } from '#lib/nntp/index' interface Target { @@ -14,6 +15,10 @@ interface Target { missing: ChunkSpan[] /** All of the range's chunks, from its start. */ total: number + start: number + end: number + /** Its chunks are kept in the corpus (ranges.capture), and read from there when stored. */ + capture: boolean } /** A request's bytes in are sampled this often; STALL_SAMPLES of them make the minute a stall is judged over. */ @@ -69,6 +74,8 @@ export class Prefetcher { private freshStats() { return { chunks: 0, + /** Chunks read from the corpus instead of fetched. */ + stored: 0, articles: 0, fetchMs: 0, hedged: 0, @@ -141,15 +148,23 @@ export class Prefetcher { await client.group(target.group) return client.xoverRaw(span.from, span.to) } - let data: Buffer + const corpusRange = { group: target.group, start: target.start, end: target.end } + // A range in the corpus is read from disk (a reset re-indexes it without the news server). + let data = target.capture ? await Corpus.read(corpusRange, span) : null + const stored = data !== null try { - data = target.plain.has(span.from) ? await NntpPool.usePlain(fetch) : await this.fetchHedged(target.id, target.group, span) + if (!data) data = target.plain.has(span.from) ? await NntpPool.usePlain(fetch) : await this.fetchHedged(target.id, target.group, span) } finally { this.fetchingNow-- } - this.stats.chunks++ - this.stats.articles += span.to - span.from + 1 - this.stats.fetchMs += Date.now() - fetchedAt + if (stored) { + this.stats.stored++ + } else { + this.stats.chunks++ + this.stats.articles += span.to - span.from + 1 + this.stats.fetchMs += Date.now() - fetchedAt + if (target.capture) await Corpus.write(corpusRange, span, data, target.total) + } const writtenAt = Date.now() this.writingNow++ try { @@ -288,7 +303,8 @@ export class Prefetcher { logger.info( `Prefetched ${chunks} chunks in ${seconds.toFixed(0)} s: ${Math.round(articles / seconds).toLocaleString('en')} articles/s, ` + `${(fetchMs / chunks / 1000).toFixed(1)} s a chunk; ` + - `hedged ${hedged} (second won ${hedgeWon}); slow by connection age: ${ages}.` + `hedged ${hedged} (second won ${hedgeWon}); slow by connection age: ${ages}` + + (this.stats.stored > 0 ? `; ${this.stats.stored} chunks from the corpus.` : '.') ) const st = this.stats const avg = (n: number) => (samples > 0 ? n / samples : 0) @@ -344,7 +360,7 @@ export class Prefetcher { this.spoolBytes = await Spool.size() } const mine = (q: any) => (this.shard.count > 1 ? q.whereRaw('r.id % ? = ?', [this.shard.count, this.shard.index]) : q) - const rows = (q: any) => mine(q.from('ranges as r').join('groups as g', 'g.id', 'r.group_id').select('r.id', 'g.name as group', 'r.start_id', 'r.end_id', 'r.next_id', 'r.plain_chunks')) + const rows = (q: any) => mine(q.from('ranges as r').join('groups as g', 'g.id', 'r.group_id').select('r.id', 'g.name as group', 'r.start_id', 'r.end_id', 'r.next_id', 'r.plain_chunks', 'r.capture')) // 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') // Planned ranges ahead: those started already, then new ones up to SPOOL_AHEAD_PER_LEASE per lease. @@ -370,7 +386,9 @@ export class Prefetcher { 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 (missing.length > 0) targets.push({ id: Number(row.id), group: row.group, plain: new Set((row.plain_chunks ?? []).map(Number)), missing, total }) + if (missing.length > 0) { + targets.push({ id: Number(row.id), group: row.group, plain: new Set((row.plain_chunks ?? []).map(Number)), missing, total, start, end, capture: row.capture === true }) + } } return targets } diff --git a/config/coordinator.ts b/config/coordinator.ts index 6a16475..dc27def 100644 --- a/config/coordinator.ts +++ b/config/coordinator.ts @@ -75,6 +75,10 @@ export default { /** Candidates claimed at a time. */ batch: 200, }, + /** Stored XOVER chunks of ranges marked `capture`, for the grouping sweeps (app/services/corpus.ts). Off when unset. */ + corpus: { + dir: env.get('CORPUS_DIR'), + }, spool: { dir: env.get('SPOOL_DIR', app.tmpPath('spool')), /** Ranges kept fetched ahead, per active lease. */ diff --git a/database/migrations/1770000002100_ranges_capture.ts b/database/migrations/1770000002100_ranges_capture.ts new file mode 100644 index 0000000..70b27c2 --- /dev/null +++ b/database/migrations/1770000002100_ranges_capture.ts @@ -0,0 +1,12 @@ +import { BaseSchema } from '@adonisjs/lucid/schema' + +/** Ranges whose XOVER chunks are kept in the corpus (CORPUS_DIR) for the grouping sweeps. */ +export default class extends BaseSchema { + async up() { + this.schema.raw(`ALTER TABLE ranges ADD COLUMN IF NOT EXISTS capture boolean NOT NULL DEFAULT false`) + } + + async down() { + this.schema.raw(`ALTER TABLE ranges DROP COLUMN IF EXISTS capture`) + } +} diff --git a/deploy/README.md b/deploy/README.md index da20f36..1af6096 100644 --- a/deploy/README.md +++ b/deploy/README.md @@ -34,6 +34,8 @@ homelab with the deploy key, whose forced command | `src/` | the code at the deployed commit (the build context) | | `pg/` | Postgres data (HDD) | | `spool/` | prefetched XOVER chunks | +| `corpus/` | stored XOVER chunks of ranges marked `capture`, for the grouping sweeps (up to 300 GB) | +| `corpus-out/` | the sweeps' replays, per edge image (`python3 src/deploy/corpus_sweep.py`) | | `deployed.log` | one line per deploy: sha, time (local) | | `.cache/repo.git` | bare clone the deploys fetch into | | `app/` | an older copy of the code with dev dependencies, for `tools/*.ts` (`r.sh`); not what runs | diff --git a/deploy/compose.yaml b/deploy/compose.yaml index e0dd616..2f59289 100644 --- a/deploy/compose.yaml +++ b/deploy/compose.yaml @@ -19,8 +19,11 @@ x-app: &app DB_PORT: 5432 APP_URL: http://web:3334 SPOOL_DIR: /spool + # Stored XOVER chunks of the ranges marked `capture` (the grouping sweeps' corpus, HDD; app/services/corpus.ts). + CORPUS_DIR: /corpus volumes: - ./spool:/spool + - ./corpus:/corpus stop_grace_period: 60s depends_on: migrate: diff --git a/deploy/corpus_sweep.py b/deploy/corpus_sweep.py new file mode 100644 index 0000000..cff37a2 --- /dev/null +++ b/deploy/corpus_sweep.py @@ -0,0 +1,203 @@ +#!/usr/bin/env python3 +"""The grouping sweeps' corpus (REBUILD.md, corpus): stored XOVER chunks replayed with each build's rules. + +Run on homelab from the deployment root (/mnt/bulk/usenet-rebuild): + + python3 src/deploy/corpus_sweep.py capture [N] mark N more ranges to store (default 100), fetched first + python3 src/deploy/corpus_sweep.py replay edge-bench over every stored range with the deployed edge image + python3 src/deploy/corpus_sweep.py report what the replay grouped: suspects, and the change since the last build + +The prefetcher stores the chunks of ranges marked `capture` in corpus//-/ (range.json once +complete) and reads them from there afterwards, so a reset doesn't fetch them again. Replays go to +corpus-out///-.jsonl, one per build; `report` compares the newest two. +""" +import collections +import json +import os +import shutil +import subprocess +import sys +from datetime import datetime, timezone + +ROOT = os.getcwd() +CORPUS = os.path.join(ROOT, 'corpus') +OUT = os.path.join(ROOT, 'corpus-out') +IMAGE = 'usenet-rebuild-edge:latest' +# The corpus may take this much of the HDD (user, 2026-10-12: up to 300 GB). +MAX_BYTES = 300e9 +# Ranks in groups.tsv the new ranges come from first (the top 100), then the rest of what's indexed. +TOP = 100 + + +def env(name): + for line in open(os.path.join(ROOT, '.env')): + if line.startswith(name + '='): + return line.split('=', 1)[1].strip() + raise SystemExit(f'{name} not in .env') + + +def psql(sql): + """Runs SQL in the catalog; rows as lists of strings.""" + out = subprocess.run( + ['docker', 'compose', 'exec', '-T', 'postgres', 'psql', '-U', env('DB_USER'), '-d', env('DB_DATABASE'), '-At', '-F', '\t', '-v', 'ON_ERROR_STOP=1'], + input=sql, capture_output=True, text=True, check=True, + ).stdout + return [line.split('\t') for line in out.splitlines() if line] + + +def corpus_bytes(): + total = 0 + for dirpath, _, files in os.walk(CORPUS): + total += sum(os.path.getsize(os.path.join(dirpath, f)) for f in files) + return total + + +def capture(n): + used = corpus_bytes() + if used >= MAX_BYTES: + print(f'corpus at {used / 1e9:.0f} GB of {MAX_BYTES / 1e9:.0f}: nothing more marked') + return + top = [line.split('\t')[1] for line in open(os.path.join(ROOT, 'groups.tsv')).read().splitlines()[:TOP]] + names = ','.join("'" + g.replace("'", "''") + "'" for g in top) + # Spread over the groups and their history: per group, the ranges farthest from those stored already + # (at random in a group without any), the top-100 groups first. Only ranges not started (a range half + # spooled would miss its first chunks; a done one is stored at the next reset, which re-reads them all). + # Priority 0: fetched before the rest of the history (the tail stays first). + rows = psql(f""" +WITH stored AS (SELECT group_id, array_agg(start_id) AS starts FROM ranges WHERE capture GROUP BY 1), +pick AS ( + SELECT r.id, g.name IN ({names}) AS top, row_number() OVER (PARTITION BY r.group_id ORDER BY + coalesce((SELECT min(abs(r.start_id - s)) FROM unnest(st.starts) s), 0) DESC, random()) AS n + FROM ranges r JOIN groups g ON g.id = r.group_id LEFT JOIN stored st ON st.group_id = r.group_id + WHERE r.priority >= 0 AND NOT r.capture AND r.status = 'planned' AND r.spooled_chunks = 0 AND r.next_id = r.start_id +), +chosen AS (SELECT id FROM pick ORDER BY top DESC, n, random() LIMIT {int(n)}) +UPDATE ranges r SET capture = true, priority = 0 FROM chosen WHERE r.id = chosen.id RETURNING r.id; +""") + total = psql('SELECT count(*), count(*) FILTER (WHERE status = \'done\') FROM ranges WHERE capture;')[0] + print(f'marked {len(rows)} ranges; {total[0]} marked in all, {total[1]} indexed; corpus {used / 1e9:.1f} GB') + + +def complete_ranges(): + for group in sorted(os.listdir(CORPUS)) if os.path.isdir(CORPUS) else []: + for rng in sorted(os.listdir(os.path.join(CORPUS, group))): + if os.path.exists(os.path.join(CORPUS, group, rng, 'range.json')): + yield group, rng + + +def image_id(): + return subprocess.run(['docker', 'image', 'inspect', '-f', '{{.Id}}', IMAGE], capture_output=True, text=True, check=True).stdout.strip().split(':')[-1][:12] + + +def replay(): + build = image_id() + out = os.path.join(OUT, build) + todo = [(g, r) for g, r in complete_ranges() if not os.path.exists(os.path.join(out, g, r + '.jsonl'))] + for g, _ in todo: + os.makedirs(os.path.join(out, g), exist_ok=True) + print(f'build {build}: {len(todo)} ranges to replay', flush=True) + if not todo: + return + # One container, 6 at a time (the edge worker keeps its threads). + script = 'xargs -P 6 -n 2 sh -c \'edge-bench "/corpus/$0/$1" --releases "/out/$0/$1.jsonl.part" >/dev/null 2>&1 && mv "/out/$0/$1.jsonl.part" "/out/$0/$1.jsonl"\'' + started = datetime.now() + subprocess.run( + ['docker', 'run', '--rm', '-i', '--user', '1000:1000', '-v', f'{CORPUS}:/corpus:ro', '-v', f'{out}:/out', '--entrypoint', 'sh', IMAGE, '-c', script], + input=''.join(f'{g} {r}\n' for g, r in todo), text=True, check=True, + ) + print(f'replayed in {(datetime.now() - started).total_seconds():.0f} s') + # Older builds' replays beyond the last two go. + builds = sorted((os.path.getmtime(os.path.join(OUT, b)), b) for b in os.listdir(OUT)) + for _, b in builds[:-2]: + shutil.rmtree(os.path.join(OUT, b)) + + +def load(build): + """Releases of one build's replay, merged by (group, key) as the catalog does.""" + releases = {} + for dirpath, _, files in os.walk(os.path.join(OUT, build)): + for f in files: + if not f.endswith('.jsonl'): + continue + for line in open(os.path.join(dirpath, f), encoding='utf-8', errors='replace'): + r = json.loads(line) + k = (r['group'], r['key']) + have = releases.get(k) + if have is None: + releases[k] = r + else: + have['files'] += r['files'] + have['segments'] += r['segments'] + have['subjects'] = (have['subjects'] + r['subjects'])[:3] + have['fileTotal'] = have['fileTotal'] or r['fileTotal'] + return releases + + +def hour(r): + return datetime.fromtimestamp((r['postedAt'] or 0) / 1000, timezone.utc).strftime('%Y-%m-%d %H') + + +def show(r, prefix=' '): + print(f"{prefix}{r['files']} files, total {r['fileTotal']}, {r['source']}: {r['name'][:80]!r} ({r['poster'][:30]}, {r['group']})") + for s in r['subjects'][:2]: + print(f'{prefix} {s[:150]}') + + +def report(): + builds = sorted((os.path.getmtime(os.path.join(OUT, b)), b) for b in os.listdir(OUT)) if os.path.isdir(OUT) else [] + if not builds: + raise SystemExit('nothing replayed yet') + build = builds[-1][1] + rel = load(build) + ranges = sum(1 for _ in complete_ranges()) + print(f'build {build}: {len(rel):,} releases, {sum(r["files"] for r in rel.values()):,} files from {ranges} stored ranges ({corpus_bytes() / 1e9:.1f} GB)') + + print('\n== same name, poster and hour: 4+ releases (under-grouping)') + clusters = collections.defaultdict(list) + for r in rel.values(): + clusters[(r['group'], r['poster'], r['name'], hour(r))].append(r) + for (g, p, name, h), rs in sorted(clusters.items(), key=lambda kv: -len(kv[1]))[:25]: + if len(rs) < 4: + break + print(f" {len(rs)} releases, {sum(r['files'] for r in rs)} files: {name[:70]!r} ({p[:25]}, {g}, {h})") + for s in rs[0]['subjects'][:1] + rs[-1]['subjects'][:1]: + print(f' {s[:150]}') + + print('\n== single files of a counter of 5+, per group and poster (under-grouping)') + singles = collections.defaultdict(list) + for r in rel.values(): + if r['files'] == 1 and (r['fileTotal'] or 0) >= 5: + singles[(r['group'], r['poster'])].append(r) + for (g, p), rs in sorted(singles.items(), key=lambda kv: -len(kv[1]))[:15]: + if len(rs) < 20: + break + print(f' {len(rs)} single files: {p[:30]} ({g})') + for s in rs[0]['subjects'][:1] + rs[-1]['subjects'][:1]: + print(f' {s[:150]}') + + print('\n== 2000+ files, or 3x the counter (over-grouping)') + big = [r for r in rel.values() if r['files'] > 2000 or (r['fileTotal'] and r['files'] > 3 * r['fileTotal'] and r['files'] > 50)] + for r in sorted(big, key=lambda r: -r['files'])[:20]: + show(r, ' ') + + if len(builds) >= 2: + before = load(builds[-2][1]) + print(f'\n== change since build {builds[-2][1]}: {len(before):,} → {len(rel):,} releases') + names = lambda rs: collections.Counter((r['group'], r['name']) for r in rs.values()) + a, b = names(before), names(rel) + moved = sorted(set(a) | set(b), key=lambda k: -abs(a[k] - b[k]))[:20] + for k in moved: + if a[k] != b[k]: + print(f' {a[k]:>6} → {b[k]:<6} {k[1][:80]!r} ({k[0]})') + + +if __name__ == '__main__': + command = sys.argv[1] if len(sys.argv) > 1 else 'report' + if command == 'capture': + capture(int(sys.argv[2]) if len(sys.argv) > 2 else 100) + elif command == 'replay': + replay() + elif command == 'report': + report() + else: + raise SystemExit(__doc__) diff --git a/edge/src/indexer.rs b/edge/src/indexer.rs index 3193c4b..be1ca67 100644 --- a/edge/src/indexer.rs +++ b/edge/src/indexer.rs @@ -47,8 +47,12 @@ pub struct ReleaseChunk { pub readable: bool, pub has_par2: bool, pub payload: Vec, + /// The first files' subjects (at most SUBJECTS): edge-bench's --releases dump for the grouping sweeps. + pub subjects: Vec, } +pub const SUBJECTS: usize = 3; + pub struct Candidate { pub release_key: String, pub upload: usize, @@ -830,6 +834,7 @@ fn chunk_of(key: &str, files: &[Vec], upload: usize) -> ReleaseChunk { readable: chunk_files.iter().any(|f| !is_random_file(f.name)), has_par2: chunk_files.iter().any(|f| names::is_par2(f.name)), payload: encode_chunk(&chunk_files), + subjects: chunk_files.iter().take(SUBJECTS).map(|f| f.subject.to_string()).collect(), } } diff --git a/edge/src/main.rs b/edge/src/main.rs index 57fb5c7..1f6966b 100644 --- a/edge/src/main.rs +++ b/edge/src/main.rs @@ -57,6 +57,37 @@ fn opt_num(n: Option, out: &mut String) { } } +/// The --releases dump: per release chunk its key, name, rule, counter total, poster, time, counts and first subjects. +fn dump_releases(batch: &Batch, group: &str, out: &mut impl Write) { + let mut line = String::new(); + for r in &batch.releases { + line.push_str("{\"group\":"); + json_str(group, &mut line); + line.push_str(",\"key\":"); + json_str(&r.key, &mut line); + line.push_str(",\"name\":"); + json_str(&r.name, &mut line); + line.push_str(&format!(",\"source\":\"{}\",\"fileTotal\":", r.source)); + opt_num(r.file_total, &mut line); + line.push_str(",\"poster\":"); + json_str(&r.poster, &mut line); + line.push_str(",\"postedAt\":"); + opt_num(r.posted_at, &mut line); + line.push_str(",\"postedLast\":"); + opt_num(r.posted_last, &mut line); + line.push_str(&format!(",\"files\":{},\"segments\":{},\"bytes\":{},\"firstArticle\":{},\"hasPar2\":{},\"subjects\":[", r.files, r.segments, r.bytes, r.first_article, r.has_par2)); + for (i, s) in r.subjects.iter().enumerate() { + if i > 0 { + line.push(','); + } + json_str(s, &mut line); + } + line.push_str("]}\n"); + out.write_all(line.as_bytes()).unwrap(); + line.clear(); + } +} + fn dump_batch(batch: &Batch, out: &mut impl Write) { let b64 = base64::engine::general_purpose::STANDARD; let mut line = String::new(); @@ -157,6 +188,8 @@ fn main() { 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())); + // 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"); @@ -214,6 +247,9 @@ fn main() { 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); + } } let total = inflate_s + index_s; #[allow(static_mut_refs)] diff --git a/start/env.ts b/start/env.ts index 98c8104..5bada5a 100644 --- a/start/env.ts +++ b/start/env.ts @@ -60,6 +60,7 @@ export default await Env.create(new URL('../', import.meta.url), { CHUNK_ARTICLES: Env.schema.number.optional(), LEASE_TTL_SECONDS: Env.schema.number.optional(), SPOOL_DIR: Env.schema.string.optional(), + CORPUS_DIR: Env.schema.string.optional(), SPOOL_AHEAD_PER_LEASE: Env.schema.number.optional(), SPOOL_MIN_AHEAD: Env.schema.number.optional(), SPOOL_MAX_GB: Env.schema.number.optional(),