Compare commits
2
Commits
d773c91ad0
...
a18ead666b
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
a18ead666b | ||
|
|
7dc316c987 |
No files matched your search
+5
-2
@@ -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
|
||||||
|
|||||||
@@ -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`)
|
||||||
|
}
|
||||||
|
}
|
||||||
@@ -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'
|
||||||
@@ -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,
|
||||||
|
}
|
||||||
|
}
|
||||||
@@ -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(),
|
||||||
})
|
})
|
||||||
@@ -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)
|
||||||
|
})
|
||||||
|
})
|
||||||
Reference in new issue
Block a user