diff --git a/REBUILD.md b/REBUILD.md index d275b5a..8c44f10 100644 --- a/REBUILD.md +++ b/REBUILD.md @@ -230,6 +230,20 @@ 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. +Since 2026-10-23 also history fetched ahead of the indexing: while the +catalog is over its NVMe cap, ranges marked `capture` (priority >= 0, not +leased) are fetched into the corpus by idle loops, at most perGroup / +perRange of a group at a time, to be indexed from disk once there's room +(LOG.md 2026-10-23 (100)). Stored filtered (`CORPUS_FILTER`, default on): +without the lines the edge worker only counts, decided per line as it does +(a crosspost another indexed group owns, a subject that names no file part; +~48% of headers), the chunk compressed again as zlib; beside it a `.drops` +(gzip JSON: rule version, the owners' priorities, per rule the article +numbers as differences, crossposts by owner) and `filtered` in range.json. +The worker makes the same releases of a filtered chunk; rule_counts lack +those two rules' counts. `corpus:filter` filters chunks stored raw. A +corpus filtered this way no longer serves a sweep of those two rules. + ## 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 index bb636b8..ca8ca62 100644 --- a/app/services/corpus.ts +++ b/app/services/corpus.ts @@ -1,11 +1,22 @@ import { access, mkdir, readFile, readdir, rename, rm, writeFile } from 'node:fs/promises' import { join } from 'node:path' -import { unzip } from 'node:zlib' +import { deflate as deflateCb, gzip as gzipCb, unzip } from 'node:zlib' import { promisify } from 'node:util' import coordinatorConfig from '#config/coordinator' +import { filterOverviews } from '#lib/edge/index' import type { ChunkSpan } from '#services/spool' const inflate = promisify(unzip) +const deflate = promisify(deflateCb) +const gzip = promisify(gzipCb) + +/** What a filtered chunk was filtered with (CORPUS_FILTER): the rule set and the group priorities of its lease. */ +export interface CorpusFilter { + ruleVersion: number + priorities: Record +} +/** The rules a filtered chunk leaves out (lib/edge/corpus_filter.ts), in its range.json. */ +export const FILTERED = ['crosspost', 'obfuscated'] export interface CorpusRange { group: string @@ -19,7 +30,10 @@ export interface CorpusRange { * 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. + * again. Off without CORPUS_DIR. Filtered (CORPUS_FILTER; LOG.md 2026-10-23 + * (101)): the lines the edge worker only counts are left out, ~half of them, + * and each chunk's `.drops` beside it (gzip JSON, ChunkDrops) lists their + * article numbers by rule; its range.json says `filtered`. */ class Corpus { get dir(): string | undefined { @@ -61,28 +75,40 @@ class Corpus { /** * Stores a chunk that inflates (a reply cut short isn't kept), and the - * range.json once the range is complete (`total` chunks). + * range.json once the range is complete (`total` chunks). With `filter`, + * without the lines the edge worker drops on their own, compressed again + * as the news server's (zlib), with its `.drops` written first. */ - async write(range: CorpusRange, span: ChunkSpan, data: Buffer, total: number) { + async write(range: CorpusRange, span: ChunkSpan, data: Buffer, total: number, filter?: CorpusFilter) { if (!this.dir) return + let raw: Buffer try { - await inflate(data) + raw = 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) + let stored = data + if (filter) { + // Latin-1 both ways: every byte back as it was. + const { kept, drops } = filterOverviews(raw.toString('latin1'), range.group, filter.priorities, filter.ruleVersion) + stored = await deflate(Buffer.from(kept, 'latin1'), { level: 9 }) + await writeFile(`${file}.drops.part`, await gzip(JSON.stringify(drops))) + await rename(`${file}.drops.part`, `${file}.drops`) + } + await writeFile(`${file}.part`, stored) 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)) + if (chunks >= total) await writeFile(join(dir, 'range.json'), JSON.stringify(filter ? { ...range, filtered: FILTERED } : 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(`${this.path(range, span)}.drops`, { force: true }) await rm(join(this.rangeDir(range), 'range.json'), { force: true }) } } diff --git a/app/services/prefetcher.ts b/app/services/prefetcher.ts index 7e3a26a..8cc6f2b 100644 --- a/app/services/prefetcher.ts +++ b/app/services/prefetcher.ts @@ -3,7 +3,8 @@ 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 Corpus, { type CorpusFilter } from '#services/corpus' +import RuleSets from '#services/rule_sets' import CatalogSpace from '#services/catalog_space' import { hedge, type Request, type XoverClient } from '#lib/nntp/index' @@ -175,7 +176,7 @@ export class Prefetcher { 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) + if (target.capture) await Corpus.write(corpusRange, span, data, target.total, await this.corpusFilter()) } if (target.corpusOnly) { this.stats.captured++ @@ -415,6 +416,18 @@ export class Prefetcher { return targets } + private filterRules: { at: number; filter: CorpusFilter } | null = null + + /** The rules corpus chunks are filtered with (CORPUS_FILTER), read again once a minute; undefined when off. */ + private async corpusFilter(): Promise { + if (!coordinatorConfig.corpus.filter) return undefined + if (!this.filterRules || Date.now() - this.filterRules.at > 60_000) { + const rules = await RuleSets.current() + this.filterRules = { at: Date.now(), filter: { ruleVersion: rules.version, priorities: rules.priorities } } + } + return this.filterRules.filter + } + /** * Up to corpus.ahead planned capture ranges, not leased soon, whose chunks aren't all in the corpus; as many of a * group as its connections allow (perGroup / perRange): in id order the 16 were one group's, which took 4 of the diff --git a/commands/corpus_filter.ts b/commands/corpus_filter.ts new file mode 100644 index 0000000..f7c319a --- /dev/null +++ b/commands/corpus_filter.ts @@ -0,0 +1,54 @@ +import { BaseCommand, flags } from '@adonisjs/core/ace' +import type { CommandOptions } from '@adonisjs/core/types/ace' +import { access, readFile, readdir, writeFile } from 'node:fs/promises' +import { join } from 'node:path' +import db from '@adonisjs/lucid/services/db' +import Corpus, { FILTERED } from '#services/corpus' +import RuleSets from '#services/rule_sets' +import coordinatorConfig from '#config/coordinator' + +/** + * Filters corpus chunks stored as the news server sent them (before CORPUS_FILTER, 2026-10-23): each chunk without a + * `.drops` of the planned capture ranges (--all: of every range in the corpus) written again as Corpus.write would. + */ +export default class CorpusFilterCommand extends BaseCommand { + static commandName = 'corpus:filter' + static description = 'Filters stored corpus chunks (planned capture ranges; --all: every range) as CORPUS_FILTER does' + static options: CommandOptions = { startApp: true } + + @flags.boolean({ description: 'Every range in the corpus, also done ones (the grouping sweeps read them raw)' }) + declare all: boolean + + async run() { + const dir = coordinatorConfig.corpus.dir + if (!dir) return this.logger.error('CORPUS_DIR is not set') + const rules = await RuleSets.current() + const filter = { ruleVersion: rules.version, priorities: rules.priorities } + const query = db.from('ranges as r').join('groups as g', 'g.id', 'r.group_id').where('r.capture', true).select('g.name as group', 'r.start_id', 'r.end_id') + const ranges = await (this.all ? query : query.where('r.status', 'planned')) + let chunks = 0 + let before = 0 + let after = 0 + for (const r of ranges) { + const range = { group: r.group, start: Number(r.start_id), end: Number(r.end_id) } + const rangeDir = join(dir, range.group, `${range.start}-${range.end}`) + const files = await readdir(rangeDir).catch(() => [] as string[]) + const gz = files.filter((f) => f.endsWith('.gz')) + for (const name of gz) { + if (files.includes(`${name}.drops`)) continue + const m = /^\d{4}-(\d+)-(\d+)\.gz$/.exec(name) + if (!m) continue + const data = await readFile(join(rangeDir, name)) + await Corpus.write(range, { from: Number(m[1]), to: Number(m[2]) }, data, Number.POSITIVE_INFINITY, filter) + before += data.length + after += (await readFile(join(rangeDir, name))).length + chunks++ + } + // A complete range's range.json says it's filtered now (Corpus.write only writes it at the last chunk). + const json = join(rangeDir, 'range.json') + if (gz.length > 0 && (await access(json).then(() => true, () => false))) await writeFile(json, JSON.stringify({ ...range, filtered: FILTERED })) + if (chunks > 0 && chunks % 1000 < gz.length) this.logger.info(`${chunks} chunks filtered: ${(before / 1e9).toFixed(1)} → ${(after / 1e9).toFixed(1)} GB.`) + } + this.logger.info(`Done: ${chunks} chunks of ${ranges.length} ranges filtered, ${(before / 1e9).toFixed(1)} → ${(after / 1e9).toFixed(1)} GB.`) + } +} diff --git a/config/coordinator.ts b/config/coordinator.ts index 3717ccd..b4f25af 100644 --- a/config/coordinator.ts +++ b/config/coordinator.ts @@ -131,6 +131,12 @@ export default { * ranges waited behind it for hours (1,400 marked, none stored, 34 of 40 connections idle; LOG.md 2026-10-12 (27)). */ ahead: env.get('CORPUS_AHEAD', 16), + /** + * Stored without the lines the edge worker drops on their own (crossposts another indexed group owns, obfuscated + * subjects; ~48% of the indexed ranges' headers), what was left out in each chunk's `.drops`: twice the history + * in the same space (LOG.md 2026-10-23 (101)). Off: as the news server sent them. + */ + filter: env.get('CORPUS_FILTER', true), }, spool: { dir: env.get('SPOOL_DIR', app.tmpPath('spool')), diff --git a/lib/edge/corpus_filter.ts b/lib/edge/corpus_filter.ts new file mode 100644 index 0000000..fea558c --- /dev/null +++ b/lib/edge/corpus_filter.ts @@ -0,0 +1,78 @@ +import { SegmentScanner, scanOverviewText } from './scan.js' + +/** + * What a corpus chunk left out (REBUILD.md, corpus; LOG.md 2026-10-23 (101)): the overview lines the edge worker only + * counts and never indexes, decided per line as RangeIndexer.line does: a crosspost an indexed group with a lower + * priority number owns (by owner), and a subject that names no file part (obfuscated). Article numbers ascending, + * stored as differences. Release-level rules (garbage, per-article, executable, poster) judge files put together + * across chunks, so they aren't applied here. + */ +export interface ChunkDrops { + version: 1 + /** The rule set (rule_sets.version) and group priorities the decision was made with. */ + ruleVersion: number + priorities: Record + /** Overview lines in the chunk, and kept. */ + lines: number + kept: number + /** By owning group, article numbers as differences (the first from 0). */ + crosspost: Record + obfuscated: number[] +} + +const deltas = (numbers: number[]) => numbers.map((n, i) => (i === 0 ? n : n - numbers[i - 1])) + +/** The article numbers back from ChunkDrops differences. */ +export function undelta(differences: number[]): number[] { + let at = 0 + return differences.map((d) => (at += d)) +} + +/** + * A chunk's overview text without the lines the edge worker drops on their own, and what was dropped. Lines kept + * byte for byte (with their line ends); lines the worker skips uncounted (no number or message-id) are kept. + */ +export function filterOverviews(text: string, group: string, priorities: Record, ruleVersion: number): { kept: string; drops: ChunkDrops } { + const scanner = new SegmentScanner() + const crosspost = new Map() + const obfuscated: number[] = [] + const dropped = new Set() + let lines = 0 + scanOverviewText(text, (n, subject, from, date, messageId, bytes, xref) => { + lines++ + if (!Number.isFinite(n) || !messageId) return + const owner = xref ? scanner.owner(xref, group, priorities) : group + if (owner !== group) { + const list = crosspost.get(owner) + if (list) list.push(n) + else crosspost.set(owner, [n]) + dropped.add(n) + } else if (!scanner.segment(n, subject, from, date, messageId, bytes, xref, group)) { + obfuscated.push(n) + dropped.add(n) + } + }) + const out: string[] = [] + let start = 0 + while (start < text.length) { + let end = text.indexOf('\n', start) + end = end === -1 ? text.length : end + 1 + const tab = text.indexOf('\t', start) + const n = tab !== -1 && tab < end ? Number.parseInt(text.slice(start, tab), 10) : Number.NaN + if (!dropped.has(n)) out.push(text.slice(start, end)) + start = end + } + const sorted = (list: number[]) => [...list].sort((a, b) => a - b) + return { + kept: out.join(''), + drops: { + version: 1, + ruleVersion, + priorities: Object.fromEntries(Object.entries(priorities).filter(([name]) => crosspost.has(name) || name === group)), + lines, + kept: lines - dropped.size, + crosspost: Object.fromEntries([...crosspost].map(([owner, list]) => [owner, deltas(sorted(list))])), + obfuscated: deltas(sorted(obfuscated)), + }, + } +} diff --git a/lib/edge/index.ts b/lib/edge/index.ts index 4b68a9c..9adc6af 100644 --- a/lib/edge/index.ts +++ b/lib/edge/index.ts @@ -5,3 +5,4 @@ export type { Lease } from './lease.js' export { mergeFragments } from './merge.js' export type { EdgeSegment, FileMeta } from './scan.js' export { throttle } from './throttle.js' +export { filterOverviews, undelta, type ChunkDrops } from './corpus_filter.js' diff --git a/start/env.ts b/start/env.ts index 6db9cf2..7c74004 100644 --- a/start/env.ts +++ b/start/env.ts @@ -51,6 +51,7 @@ export default await Env.create(new URL('../', import.meta.url), { NNTP_HEDGE_CONNECTIONS: Env.schema.number.optional(), NNTP_CONNECTION_MAX_MINUTES: Env.schema.number.optional(), NNTP_IDLE_CLOSE_SECONDS: Env.schema.number.optional(), + CORPUS_FILTER: Env.schema.boolean.optional(), NNTP_STALL_KB_PER_MINUTE: Env.schema.number.optional(), // Newznab API key (empty: no key needed) diff --git a/tests/unit/edge/corpus_filter.spec.ts b/tests/unit/edge/corpus_filter.spec.ts new file mode 100644 index 0000000..cd6b419 --- /dev/null +++ b/tests/unit/edge/corpus_filter.spec.ts @@ -0,0 +1,83 @@ +import { test } from '@japa/runner' +import { mkdtemp, readFile, rm } from 'node:fs/promises' +import { tmpdir } from 'node:os' +import { join } from 'node:path' +import { deflateSync, gunzipSync, inflateSync } from 'node:zlib' +import coordinatorConfig from '#config/coordinator' +import Corpus from '#services/corpus' +import { RangeIndexer, filterOverviews, undelta } from '#lib/edge/index' +import { DEFAULT_RULES } from '#lib/rules/index' + +const GROUP = 'alt.binaries.test' +const priorities = { 'alt.binaries.boneless': 1, [GROUP]: 2, 'alt.binaries.x': 3 } + +/** An overview line as XOVER sends it (number, subject, from, date, message-id, references, bytes, lines, Xref). */ +const line = (n: number, subject: string, xref = `news.example ${GROUP}:${n}`, messageId = ``) => + [n, subject, 'poster ', 'Mon, 05 Oct 2026 10:00:00 +0000', messageId, '', '700000', '5000', `Xref: ${xref}`].join('\t') + +const LINES = [ + line(100, 'Some.Show.S01E01 [1/2] - "Some.Show.S01E01.1080p.part1.rar" yEnc (1/2)'), + // Owned by boneless (priority 1): the worker counts it there. + line(101, 'Other [1/1] - "Other.mkv" yEnc (1/1)', `news.example ${GROUP}:101 alt.binaries.boneless:9`), + line(102, 'Some.Show.S01E01 [1/2] - "Some.Show.S01E01.1080p.part1.rar" yEnc (2/2)'), + // No file part in the subject. + line(103, 'obfuscated'), + // Also in a lower-priority group (3): ours. + line(104, 'Some.Show.S01E01 [2/2] - "Some.Show.S01E01.1080p.par2" yEnc (1/1)', `news.example ${GROUP}:104 alt.binaries.x:7`), + line(105, 'Pq4ZmXc8VbN2LkJh6GfDs3'), + // Without a message-id: the worker skips it uncounted, so it stays. + line(106, 'x', undefined, ''), +] +const TEXT = LINES.map((l) => `${l}\r\n`).join('') + +function index(text: string) { + const indexer = new RangeIndexer({ group: GROUP, start: 1, end: 1000, rules: { ...DEFAULT_RULES, samplePercent: 0, droppedSignatures: [], priorities }, edgeArticles: 10 }) + indexer.addText(text, 1001) + indexer.finish() + return indexer.take() +} + +test.group('edge / corpus filter', () => { + test('leaves out crossposts another group owns and obfuscated subjects, keeps the rest byte for byte', ({ assert }) => { + const { kept, drops } = filterOverviews(TEXT, GROUP, priorities, 7) + assert.equal(kept, [LINES[0], LINES[2], LINES[4], LINES[6]].map((l) => `${l}\r\n`).join('')) + assert.equal(drops.ruleVersion, 7) + assert.equal(drops.lines, 7) + assert.equal(drops.kept, 4) + assert.deepEqual(Object.keys(drops.crosspost), ['alt.binaries.boneless']) + assert.deepEqual(undelta(drops.crosspost['alt.binaries.boneless']), [101]) + assert.deepEqual(undelta(drops.obfuscated), [103, 105]) + assert.deepEqual(drops.priorities, { 'alt.binaries.boneless': 1, [GROUP]: 2 }) + }) + + test('the worker makes the same releases, candidates and fragments of the filtered chunk', ({ assert }) => { + const raw = index(TEXT) + const filtered = index(filterOverviews(TEXT, GROUP, priorities, 7).kept) + assert.isAbove(raw.releases.length, 0) + assert.deepEqual(filtered.releases, raw.releases) + assert.deepEqual(filtered.candidates, raw.candidates) + assert.deepEqual(filtered.fragments, raw.fragments) + // Only the counts of what was left out differ. + const without = (b: typeof raw) => b.counts.filter((c) => c.rule !== 'crosspost' && c.rule !== 'obfuscated') + assert.deepEqual(without(filtered), without(raw)) + }) + + test('the corpus stores a filtered chunk as zlib with its .drops, and says so in range.json', async ({ assert, cleanup }) => { + const dir = await mkdtemp(join(tmpdir(), 'corpus-')) + const before = coordinatorConfig.corpus.dir + coordinatorConfig.corpus.dir = dir + cleanup(async () => { + coordinatorConfig.corpus.dir = before + await rm(dir, { recursive: true, force: true }) + }) + const range = { group: GROUP, start: 1, end: 100_000 } + const span = { from: 1, to: 100_000 } + await Corpus.write(range, span, deflateSync(Buffer.from(TEXT, 'latin1')), 1, { ruleVersion: 7, priorities }) + const stored = (await Corpus.read(range, span))! + assert.equal(inflateSync(stored).toString('latin1'), filterOverviews(TEXT, GROUP, priorities, 7).kept) + const file = join(dir, GROUP, '1-100000', '0000-1-100000.gz') + const drops = JSON.parse(gunzipSync(await readFile(`${file}.drops`)).toString()) + assert.deepEqual(undelta(drops.obfuscated), [103, 105]) + assert.deepEqual(JSON.parse(await readFile(join(dir, GROUP, '1-100000', 'range.json'), 'utf8')), { ...range, filtered: ['crosspost', 'obfuscated'] }) + }) +})