From 7dc316c9874159b61ffa849db9c2415a731f915a Mon Sep 17 00:00:00 2001 From: Daan Meijer Date: Fri, 9 Oct 2026 00:10:16 +0200 Subject: [PATCH] Rebuild: the edge worker's range indexer Overview chunks in, per batch out: releases as encoded chunks, probe candidates, counts per rule (crossposts per owner group, obfuscated, garbage, per-article, incomplete), a sample of what was dropped, and fragments near the range's edges. Co-Authored-By: Claude Opus 5.5 --- lib/edge/index.ts | 1 + lib/edge/range_indexer.ts | 294 ++++++++++++++++++++++++++ tests/unit/edge/range_indexer.spec.ts | 67 ++++++ 3 files changed, 362 insertions(+) create mode 100644 lib/edge/index.ts create mode 100644 lib/edge/range_indexer.ts create mode 100644 tests/unit/edge/range_indexer.spec.ts diff --git a/lib/edge/index.ts b/lib/edge/index.ts new file mode 100644 index 0000000..d68c3b6 --- /dev/null +++ b/lib/edge/index.ts @@ -0,0 +1 @@ +export { RangeIndexer, type EdgeBatch, type ReleaseChunk, type DroppedChunk, type ProbeCandidate, type CandidateKind, type RuleCount, type Fragment, type RangeIndexerOptions } from './range_indexer.js' diff --git a/lib/edge/range_indexer.ts b/lib/edge/range_indexer.ts new file mode 100644 index 0000000..f38f593 --- /dev/null +++ b/lib/edge/range_indexer.ts @@ -0,0 +1,294 @@ +import { createHash } from 'node:crypto' +import type { Overview } from '../nntp/index.js' +import { parseXrefGroups } from '../nntp/index.js' +import { segmentFromHeader, type IngestSegment } from '../headers/index.js' +import { FileAggregator, type AggregatedFile, isArchiveStart, isLaterVolume, sevenZipVolume } from '../files/index.js' +import { RuleSet, isRandomFile, ownerOf, type Rules } from '../rules/index.js' +import { encodeChunk, type ChunkFile } from '../chunks/index.js' + +/** A release's files from one batch (REBUILD.md, chunk store): what the catalog adds up, and the encoded chunk. */ +export interface ReleaseChunk { + key: string + name: string + source: string + fileTotal: number | null + poster: string + groups: string[] + postedAt: number | null + files: number + segments: number + bytes: number + firstArticle: number + lastArticle: number + payload: Buffer +} + +export interface DroppedChunk extends ReleaseChunk { + rule: string + ruleKey: string +} + +export type CandidateKind = 'archive' | 'par2' | 'lone' | '7z-tail' + +/** An article phase 2 should read (REBUILD.md, probe queue). */ +export interface ProbeCandidate { + releaseKey: string + kind: CandidateKind + messageId: string + file: string +} + +/** What a rule dropped in a batch (REBUILD.md, rules and recovery). */ +export interface RuleCount { + rule: 'crosspost' | 'obfuscated' | 'garbage' | 'per-article' | 'incomplete' + key: string + articles: number + files: number +} + +/** A segment of a file cut off by the range's edges, for the neighbouring range. */ +export interface Fragment { + fileKey: string + name: string + poster: string + subject: string + totalParts: number + groups: string[] + release: { key: string; name: string; source: string; fileTotal?: number } | null + number: number + messageId: string + bytes: number + articleNumber: number + postedAt: number | null +} + +export interface EdgeBatch { + /** First article number not covered yet. */ + next: number + headers: number + releases: ReleaseChunk[] + samples: DroppedChunk[] + candidates: ProbeCandidate[] + counts: RuleCount[] + fragments: Fragment[] +} + +export interface RangeIndexerOptions { + group: string + /** Inclusive article numbers. */ + start: number + end: number + rules: Rules + /** Chunks without a new part before an open file is given up (as a fragment near an edge). */ + idleChunks?: number + /** Within this many articles of an edge, a cut-off file is kept as fragments. */ + edgeArticles?: number +} + +const MAX_CANDIDATES = 4 + +/** + * One leased range on an edge worker (REBUILD.md, part 2): overview lines + * in, chunk by chunk and in order; releases, probe candidates, rule counts + * and edge fragments out, per batch (`take`). Holds only the files still + * waiting for parts. + */ +export class RangeIndexer { + private aggregator = new FileAggregator() + private rules: RuleSet + private batch: EdgeBatch + private counts = new Map() + private readonly idle: number + private readonly edge: number + + constructor(private options: RangeIndexerOptions) { + this.rules = new RuleSet(options.rules) + this.idle = options.idleChunks ?? 5 + this.edge = options.edgeArticles ?? 50_000 + this.batch = this.empty(options.start) + } + + /** One chunk of overview lines; `next` is the first article after it. */ + add(overviews: Overview[], next: number) { + const { group, rules } = this.options + const segments: IngestSegment[] = [] + for (const [articleNumber, header] of overviews) { + if (!Number.isFinite(articleNumber) || !header['message-id']) continue + this.batch.headers++ + const groups = parseXrefGroups(header.xref) + const owner = ownerOf(groups.length > 0 ? groups : [group], rules.priorities) ?? group + if (owner !== group) { + this.count('crosspost', owner, 1, 0) + continue + } + const segment = segmentFromHeader(header, group, articleNumber) + if (segment) segments.push(segment) + else this.count('obfuscated', '', 1, 0) + } + this.files(this.aggregator.add(segments)) + this.cutOff(this.aggregator.evict(this.idle)) + this.batch.next = next + } + + /** The end of the range: every open file is given up. */ + finish() { + this.cutOff(this.aggregator.evict(0)) + this.batch.next = this.options.end + 1 + } + + /** What was found since the last take, and a fresh batch. */ + take(): EdgeBatch { + const out = this.batch + out.counts = [...this.counts.values()] + this.counts.clear() + this.batch = this.empty(out.next) + return out + } + + get openFiles() { + return this.aggregator.openFiles + } + + private files(files: AggregatedFile[]) { + const byRelease = new Map[]>() + for (const file of files) { + const key = releaseKeyOf(file.segments[0]) + const list = byRelease.get(key) + if (list) list.push(file) + else byRelease.set(key, [file]) + } + for (const [key, list] of byRelease) { + const verdict = this.rules.judge( + key, + list.map((f) => ({ name: f.segments[0].name, totalParts: f.segments[0].totalParts, groups: f.segments[0].groups, firstMessageId: f.segments[0].messageId })) + ) + if (verdict.rule === null) { + this.batch.releases.push(chunkOf(key, list)) + this.batch.candidates.push(...candidatesOf(key, list)) + continue + } + this.count(verdict.rule, verdict.key, list.reduce((n, f) => n + f.segments.length, 0), list.length) + if (verdict.sample) this.batch.samples.push({ ...chunkOf(key, list), rule: verdict.rule, ruleKey: verdict.key }) + } + } + + /** Given-up segments: kept near the range's edges (the neighbour may complete them), counted otherwise. */ + private cutOff(segments: IngestSegment[]) { + const { start, end } = this.options + for (const s of segments) { + const n = Number(s.articleNumber) + if (Number.isFinite(n) && (n < start + this.edge || n > end - this.edge)) this.batch.fragments.push(fragmentOf(s)) + else this.count('incomplete', '', 1, 0) + } + } + + private count(rule: RuleCount['rule'], key: string, articles: number, files: number) { + const id = `${rule}\n${key}` + const c = this.counts.get(id) + if (c) { + c.articles += articles + c.files += files + } else { + this.counts.set(id, { rule, key, articles, files }) + } + } + + private empty(next: number): EdgeBatch { + return { next, headers: 0, releases: [], samples: [], candidates: [], counts: [], fragments: [] } + } +} + +/** The release key; a file without a recognisable release becomes a release of its own. */ +function releaseKeyOf(s: IngestSegment): string { + return s.release?.key ?? createHash('sha1').update(`file\n${s.fileKey}`).digest('hex') +} + +function postedOf(segments: IngestSegment[]): number | null { + let min: number | null = null + for (const s of segments) { + const t = s.postedAt?.getTime() + if (t !== undefined && Number.isFinite(t) && (min === null || t < min)) min = t + } + return min +} + +function chunkOf(key: string, files: AggregatedFile[]): ReleaseChunk { + const first = files[0].segments[0] + const chunkFiles: ChunkFile[] = files.map((f) => { + const s = f.segments[0] + return { name: s.name, poster: s.poster, subject: s.subject, postedAt: postedOf(f.segments), groups: s.groups, segments: f.segments.map((x) => [x.number, x.messageId, x.bytes]) } + }) + let segments = 0 + let bytes = 0 + let firstArticle = Number.POSITIVE_INFINITY + let lastArticle = 0 + let postedAt: number | null = null + for (const f of files) { + for (const s of f.segments) { + segments++ + bytes += s.bytes + const n = Number(s.articleNumber) + if (n < firstArticle) firstArticle = n + if (n > lastArticle) lastArticle = n + } + const t = postedOf(f.segments) + if (t !== null && (postedAt === null || t < postedAt)) postedAt = t + } + return { + key, + name: first.release?.name ?? first.name, + source: first.release?.source ?? 'file', + fileTotal: files.find((f) => f.segments[0].release?.fileTotal)?.segments[0].release?.fileTotal ?? null, + poster: first.poster, + groups: first.groups, + postedAt, + files: files.length, + segments, + bytes, + firstArticle, + lastArticle, + payload: encodeChunk(chunkFiles), + } +} + +/** + * The articles that say what a release is: the first of its first archive + * volume, of its par2 index, of a lone random-named file, and a 7z set's + * last volume (its listing is in the tail). At most a few per release. + */ +function candidatesOf(key: string, files: AggregatedFile[]): ProbeCandidate[] { + const out: ProbeCandidate[] = [] + let tail: { volume: number; file: AggregatedFile } | null = null + for (const f of files) { + const name = f.segments[0].name + const volume = sevenZipVolume(name) + if (volume !== null && volume > 1) { + if (!tail || volume > tail.volume) tail = { volume, file: f } + continue + } + if (isArchiveStart(name)) out.push({ releaseKey: key, kind: 'archive', messageId: f.segments[0].messageId, file: name }) + else if (/\.par2$/i.test(name) && !isLaterVolume(name)) out.push({ releaseKey: key, kind: 'par2', messageId: f.segments[0].messageId, file: name }) + } + if (tail) out.push({ releaseKey: key, kind: '7z-tail', messageId: tail.file.segments.at(-1)!.messageId, file: tail.file.segments[0].name }) + if (out.length === 0 && files.length === 1 && isRandomFile(files[0].segments[0].name)) { + out.push({ releaseKey: key, kind: 'lone', messageId: files[0].segments[0].messageId, file: files[0].segments[0].name }) + } + return out.slice(0, MAX_CANDIDATES) +} + +function fragmentOf(s: IngestSegment): Fragment { + return { + fileKey: s.fileKey, + name: s.name, + poster: s.poster, + subject: s.subject, + totalParts: s.totalParts, + groups: s.groups, + release: s.release ? { key: s.release.key, name: s.release.name, source: s.release.source, ...(s.release.fileTotal ? { fileTotal: s.release.fileTotal } : {}) } : null, + number: s.number, + messageId: s.messageId, + bytes: s.bytes, + articleNumber: Number(s.articleNumber), + postedAt: s.postedAt && Number.isFinite(s.postedAt.getTime()) ? s.postedAt.getTime() : null, + } +} diff --git a/tests/unit/edge/range_indexer.spec.ts b/tests/unit/edge/range_indexer.spec.ts new file mode 100644 index 0000000..bc2281c --- /dev/null +++ b/tests/unit/edge/range_indexer.spec.ts @@ -0,0 +1,67 @@ +import { test } from '@japa/runner' +import { RangeIndexer } from '#lib/edge/index' +import { decodeChunk } from '#lib/chunks/index' +import { DEFAULT_RULES } from '#lib/rules/index' +import type { Overview } from '#lib/nntp/index' + +const GROUP = 'alt.binaries.test' +const rules = { ...DEFAULT_RULES, samplePercent: 0, droppedSignatures: [], priorities: { 'alt.binaries.boneless': 1, [GROUP]: 2, 'alt.binaries.x': 3 } } +let id = 0 + +/** Overview lines of one posted file: all its parts, from article number `at` on. */ +function post(at: number, name: string, parts: number, opts: { counter?: string; groups?: string[]; tool?: 'ngPost' | 'nyuu'; poster?: string } = {}): Overview[] { + return Array.from({ length: parts }, (_, i) => { + id++ + const mid = opts.tool === 'ngPost' ? `${id.toString(16).padStart(32, '0')}@ngPost` : `art${id}@test` + const groups = opts.groups ?? [GROUP] + return [at + i, { + 'subject': `${opts.counter ? `${opts.counter} - ` : ''}"${name}" yEnc (${i + 1}/${parts})`, + 'from': opts.poster ?? 'poster ', + 'date': 'Mon, 05 Oct 2026 10:00:00 +0000', + 'message-id': `<${mid}>`, + ':bytes': '700000', + 'xref': `news.example ${groups.map((g, k) => `${g}:${at + i + k}`).join(' ')}`, + }] as Overview + }) +} + +test.group('edge / RangeIndexer', () => { + test('groups files into releases with probe candidates, and counts what the rules dropped', ({ assert }) => { + const indexer = new RangeIndexer({ group: GROUP, start: 1, end: 1_000_000, rules, edgeArticles: 10 }) + indexer.add([ + ...post(100_000, 'Some.Show.S01E01.1080p.WEB.part1.rar', 3, { counter: 'Some.Show.S01E01 [1/3]' }), + ...post(100_010, 'Some.Show.S01E01.1080p.WEB.part2.rar', 3, { counter: 'Some.Show.S01E01 [2/3]' }), + ...post(100_020, 'Some.Show.S01E01.1080p.WEB.par2', 1, { counter: 'Some.Show.S01E01 [3/3]' }), + // A crosspost whose owner is boneless (priority 1): skipped here. + ...post(100_030, 'Other.Release.2020.mkv', 2, { groups: [GROUP, 'alt.binaries.boneless'] }), + // Random-named ngPost upload: garbage everywhere. + ...post(100_040, 'Hs7dKq2LmZ9xWv3RtYp1Qa.part01.rar', 4, { tool: 'ngPost' }), + // A random subject per article: per-article. + ...post(100_050, 'Pq4ZmXc8VbN2LkJh6GfDs3', 1), + ], 200_000) + indexer.add([[100_060, { 'subject': 'obfuscated', 'from': 'x', 'message-id': '', 'date': '', ':bytes': '1', 'xref': `news.example ${GROUP}:100060` }]], 300_000) + indexer.finish() + const batch = indexer.take() + + assert.equal(batch.headers, 3 + 3 + 1 + 2 + 4 + 1 + 1) + assert.lengthOf(batch.releases, 1) + const [release] = batch.releases + assert.include(release, { files: 3, segments: 7, firstArticle: 100_000, lastArticle: 100_020 }) + assert.deepEqual(decodeChunk(release.payload).map((f) => f.name).sort(), ['Some.Show.S01E01.1080p.WEB.par2', 'Some.Show.S01E01.1080p.WEB.part1.rar', 'Some.Show.S01E01.1080p.WEB.part2.rar']) + assert.deepEqual(batch.candidates.map((c) => [c.kind, c.file]).sort(), [['archive', 'Some.Show.S01E01.1080p.WEB.part1.rar'], ['par2', 'Some.Show.S01E01.1080p.WEB.par2']]) + + const counts = Object.fromEntries(batch.counts.map((c) => [`${c.rule} ${c.key}`.trim(), [c.articles, c.files]])) + assert.deepEqual(counts, { 'crosspost alt.binaries.boneless': [2, 0], 'garbage ngPost alt.binaries.test': [4, 1], 'per-article other': [1, 1], 'obfuscated': [1, 0] }) + assert.equal(batch.next, 1_000_001) + }) + + test('a file cut off near an edge leaves as fragments, one in the middle is only counted', ({ assert }) => { + const indexer = new RangeIndexer({ group: GROUP, start: 1, end: 1_000_000, rules, edgeArticles: 1000 }) + indexer.add([...post(5, 'Edge.File.2020.part1.rar', 3).slice(0, 2), ...post(500_000, 'Middle.File.2020.part1.rar', 3).slice(0, 2)], 600_000) + indexer.finish() + const batch = indexer.take() + assert.deepEqual(batch.fragments.map((f) => [f.name, f.number]), [['Edge.File.2020.part1.rar', 1], ['Edge.File.2020.part1.rar', 2]]) + assert.deepEqual(batch.counts.map((c) => [c.rule, c.articles]), [['incomplete', 2]]) + assert.lengthOf(batch.releases, 0) + }) +})