2 Commits
Author SHA1 Message Date
Daan MeijerandClaude Opus 5.5 a18ead666b Rebuild: edge:bench, and its result in the spec (TypeScript is fast enough)
Test and deploy / test (push) Successful in 19s
Test and deploy / deploy (push) Skipped
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
2026-10-09 00:17:26 +02:00
Daan MeijerandClaude Opus 5.5 7dc316c987 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 <noreply@anthropic.com>
2026-10-09 00:10:16 +02:00
6 changed files with 504 additions and 2 deletions

No files matched your search

+5 -2
View File
@@ -41,8 +41,11 @@ dropping garbage; 28 TB of compressed overview traffic. Throughput goal:
Order: oldest era first (2008–2018 holds 63% of the readable files on Order: oldest era first (2008–2018 holds 63% of the readable files on
17% of the lines), then forward. 17% of the lines), then forward.
2. **Edge worker** (node, TypeScript reusing `lib/`; Rust later only if 2. **Edge worker** (node, TypeScript reusing `lib/`). Measured (`edge:bench`,
CPU-bound). Per lease: compressed chunks of 10–50 k lines, in order → 2026-10-09): 73 k headers/s per core on an a.b.ath range from 2013,
103 k/s on a recent a.b.boneless range (2 M headers each), ~1.2 GB peak
per process; the 25 B lines/day goal needs ~3–4 cores over all nodes,
so no Rust. Per lease: compressed chunks of 10–50 k lines, in order →
parse subjects → skip crossposts outside their owner group (of the parse subjects → skip crossposts outside their owner group (of the
indexed groups in the article's Xref, the one with the lowest priority indexed groups in the article's Xref, the one with the lowest priority
number; see the group list) → group articles into files → files number; see the group list) → group articles into files → files
+130
View File
@@ -0,0 +1,130 @@
import { BaseCommand, flags } from '@adonisjs/core/ace'
import type { CommandOptions } from '@adonisjs/core/types/ace'
import { mkdir, readFile, readdir, writeFile } from 'node:fs/promises'
import { join } from 'node:path'
import { unzipSync } from 'node:zlib'
import env from '#start/env'
import { XoverClient, parseOverviews } from '#lib/nntp/index'
import { RangeIndexer } from '#lib/edge/index'
import { DEFAULT_RULES, type Rules } from '#lib/rules/index'
/**
* Measures the edge worker on a real range (REBUILD.md: TypeScript or Rust):
* --capture saves a range's compressed XOVER chunks to a directory once;
* without it the saved chunks are replayed through the range indexer, timing
* inflate, parse and indexing per chunk, with memory and output sizes.
*/
export default class EdgeBench extends BaseCommand {
static commandName = 'edge:bench'
static description = 'Measures the edge worker on a captured range'
static options: CommandOptions = { startApp: true }
@flags.string({ description: 'Directory of the captured chunks', required: true })
declare dir: string
@flags.boolean({ description: 'Capture the range first (needs NNTP_* in the environment)' })
declare capture: boolean
@flags.string({ description: 'Group to capture' })
declare group?: string
@flags.number({ description: 'First article to capture' })
declare start?: number
@flags.number({ description: 'Articles to capture', default: 2_000_000 })
declare articles: number
@flags.number({ description: 'Articles per XOVER', default: 50_000 })
declare chunk: number
@flags.number({ description: 'Chunks without a new part before an open file is given up', default: 5 })
declare idle: number
@flags.string({ description: 'Indexed groups in priority order, comma-separated (owner rule)' })
declare priorities?: string
async run() {
if (this.capture) await this.captureRange()
await this.replay()
}
private async captureRange() {
if (!this.group || this.start === undefined) throw new Error('--group and --start are needed to capture')
await mkdir(this.dir, { recursive: true })
const client = await XoverClient.connect({
host: env.get('NNTP_HOST')!,
port: env.get('NNTP_PORT')!,
user: env.get('NNTP_USER'),
password: env.get('NNTP_PASSWORD'),
secure: env.get('NNTP_SECURE', true),
compress: true,
timeoutMs: 300_000,
})
await client.group(this.group)
const end = this.start + this.articles - 1
let index = 0
for (let from = this.start; from <= end; from += this.chunk, index++) {
const to = Math.min(from + this.chunk - 1, end)
const started = Date.now()
const raw = await client.xoverRaw(from, to)
await writeFile(join(this.dir, `${String(index).padStart(4, '0')}-${from}-${to}.gz`), raw)
this.logger.info(`chunk ${index}: ${from}–${to}, ${(raw.length / 1e6).toFixed(1)} MB in ${((Date.now() - started) / 1000).toFixed(1)} s`)
}
await writeFile(join(this.dir, 'range.json'), JSON.stringify({ group: this.group, start: this.start, end }))
client.close()
}
private async replay() {
const { group, start, end } = JSON.parse(await readFile(join(this.dir, 'range.json'), 'utf8')) as { group: string; start: number; end: number }
const priorities = Object.fromEntries((this.priorities ?? group).split(',').map((g, i) => [g.trim(), i + 1]))
if (!(group in priorities)) priorities[group] = Object.keys(priorities).length + 1
const rules: Rules = { ...DEFAULT_RULES, priorities, droppedSignatures: [] }
const files = (await readdir(this.dir)).filter((f) => f.endsWith('.gz')).sort()
const indexer = new RangeIndexer({ group, start, end, rules, idleChunks: this.idle })
const t = { inflate: 0, parse: 0, index: 0 }
let compressed = 0
let text = 0
let peakRss = 0
const out = { headers: 0, releases: 0, files: 0, segments: 0, payload: 0, candidates: 0, fragments: 0, samples: 0 }
const counts = new Map<string, number>()
const cpuStart = process.cpuUsage()
const wall = Date.now()
for (const file of files) {
const raw = await readFile(join(this.dir, file))
compressed += raw.length
let at = performance.now()
const data = unzipSync(raw).toString('utf8')
t.inflate += performance.now() - at
text += data.length
at = performance.now()
const overviews = parseOverviews(data)
t.parse += performance.now() - at
at = performance.now()
const next = Number(file.split('-')[2].replace('.gz', '')) + 1
indexer.add(overviews, next)
if (file === files.at(-1)) indexer.finish()
const batch = indexer.take()
t.index += performance.now() - at
out.headers += batch.headers
out.releases += batch.releases.length
for (const r of batch.releases) {
out.files += r.files
out.segments += r.segments
out.payload += r.payload.length
}
out.candidates += batch.candidates.length
out.fragments += batch.fragments.length
out.samples += batch.samples.length
for (const c of batch.counts) counts.set(`${c.rule} ${c.key}`.trim(), (counts.get(`${c.rule} ${c.key}`.trim()) ?? 0) + c.articles)
peakRss = Math.max(peakRss, process.memoryUsage().rss)
}
const cpu = process.cpuUsage(cpuStart)
const cpuS = (cpu.user + cpu.system) / 1e6
const total = t.inflate + t.parse + t.index
this.logger.info(`${group} ${start}–${end}: ${files.length} chunks, ${(compressed / 1e6).toFixed(0)} MB compressed, ${(text / 1e6).toFixed(0)} MB text`)
this.logger.info(`${out.headers.toLocaleString('en')} headers in ${(total / 1000).toFixed(1)} s (${Math.round(out.headers / (total / 1000)).toLocaleString('en')}/s on one core; CPU ${cpuS.toFixed(1)} s, wall ${((Date.now() - wall) / 1000).toFixed(1)} s)`)
this.logger.info(` inflate ${(t.inflate / 1000).toFixed(1)} s, parse ${(t.parse / 1000).toFixed(1)} s, index ${(t.index / 1000).toFixed(1)} s; peak RSS ${(peakRss / 1e6).toFixed(0)} MB`)
this.logger.info(` kept: ${out.releases} release chunks, ${out.files} files, ${out.segments} segments, payload ${(out.payload / 1e6).toFixed(1)} MB (${(out.payload / Math.max(1, out.segments)).toFixed(1)} B/segment); ${out.candidates} candidates, ${out.fragments} fragments, ${out.samples} samples`)
for (const [k, v] of [...counts].sort((a, b) => b[1] - a[1]).slice(0, 8)) this.logger.info(` dropped ${k}: ${v.toLocaleString('en')} articles`)
}
}
+1
View File
@@ -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'
+294
View File
@@ -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<IngestSegment>()
private rules: RuleSet
private batch: EdgeBatch
private counts = new Map<string, RuleCount>()
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<IngestSegment>[]) {
const byRelease = new Map<string, AggregatedFile<IngestSegment>[]>()
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<IngestSegment>[]): 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<IngestSegment>[]): ProbeCandidate[] {
const out: ProbeCandidate[] = []
let tail: { volume: number; file: AggregatedFile<IngestSegment> } | 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,
}
}
+7
View File
@@ -34,4 +34,11 @@ export default await Env.create(new URL('../', import.meta.url), {
DB_POOL_MAX: Env.schema.number.optional(), DB_POOL_MAX: Env.schema.number.optional(),
DB_SEARCH_PATH: Env.schema.string.optional(), DB_SEARCH_PATH: Env.schema.string.optional(),
DB_DEBUG: Env.schema.boolean.optional(), DB_DEBUG: Env.schema.boolean.optional(),
// NNTP (the coordinator's relay; edge:bench --capture)
NNTP_HOST: Env.schema.string.optional(),
NNTP_PORT: Env.schema.number.optional(),
NNTP_USER: Env.schema.string.optional(),
NNTP_PASSWORD: Env.schema.string.optional(),
NNTP_SECURE: Env.schema.boolean.optional(),
}) })
+67
View File
@@ -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 <p@x>',
'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': '<obf@test>', '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)
})
})