diff --git a/REBUILD.md b/REBUILD.md index d58c1e5..0a68e0d 100644 --- a/REBUILD.md +++ b/REBUILD.md @@ -254,10 +254,24 @@ with 68,032 releases with a password): | Complete RAR listings by walking volume headers | D40 | no: the prober reads the first volume's first article only | | 7z listings | D19 | no: `7z-tail` candidates are queued, not probed | | Names from yEnc bodies for obfuscated subjects | D13, D21, D36 | no: obfuscated and per-article subjects are only counted (part 5); their message-ids aren't stored, so a body path needs those ranges read again | -| Garbage dropped after probing (no real name, no password) | D42 (its second filter) | not yet; the header-time garbage rules are in (part 5) | +| Garbage dropped after probing (no real name, no password) | D42 (its second filter) | the junk judge hides it (below); nothing dropped yet | Live indexing (the tail) runs since 2026-10-09. +**Junk** (voyager 1's second filter, D42). After the header pass's rules +(garbage signatures, executables, per-article and obfuscated subjects), the +judge (`releases:judge`, service `judge`) judges releases hourly by what the +prober, the merges and the spots found, in this order (the first rule +wins): *dead* (every probed first article gone), *fake* (a film, episode +or music name whose content is a program: archive entries all +executables, or the file itself one by name or by its first bytes, sniffed +`exe`), *repeated* (the same first 16 KiB and size in 20+ releases), +*locked* (passworded, no password from a spot, its NZB or the title), +*unusable* (random names, no inner name, no par2 or inner-name merge, no +spot, a day after settling). Verdicts (`release_verdicts`, with the rule +set's version) hide releases from search (`judged` shows them); a verdict +whose rule no longer matches is taken back. Nothing is dropped yet. + **Spots** (after voyager 1's D29; `lib/spotnet` ported as is). Two processes, one NNTP connection each: diff --git a/app/services/release_judge.ts b/app/services/release_judge.ts new file mode 100644 index 0000000..2d54c83 --- /dev/null +++ b/app/services/release_judge.ts @@ -0,0 +1,142 @@ +import db from '@adonisjs/lucid/services/db' +import logger from '@adonisjs/core/services/logger' + +/** Raised when a rule changes: verdicts of an older version are judged again. */ +export const JUDGE_VERSION = 1 +/** Between rounds: each rule is a scan of its table. */ +const ROUND_MS = 60 * 60 * 1000 +/** A file's first 16 KiB and size in this many releases is one file posted again and again (test files, spam). */ +const REPEATED_RELEASES = 20 + +/** A name that says it's a film, an episode or music: a resolution, a codec, a rip source, a media extension. */ +const MEDIA = + `(^|[^a-z0-9])(s[0-9]{1,2}e[0-9]{1,3}|[0-9]{3,4}p|x26[45]|h[ .]?26[45]|hevc|xvid|divx|bluray|bdrip|brrip|web[ .-]?dl|webrip|hdtv|dvdrip|dvdr|remux|mkv|mp4|avi|wmv|m4v|flac|mp3)([^a-z0-9]|$)` +/** An executable's name. */ +const EXECUTABLE = `\\.(exe|scr|com|bat|cmd|msi|lnk|vbs|js|jar|pif|cpl|hta|ps1|dll)$` + +export interface Rule { + name: string + /** Why, in a few words (the log, the web interface). */ + says: string + /** SELECT of the release ids the rule judges. */ + sql: string +} + +/** + * The junk judge's rules, in order: a release takes the first that matches (REBUILD.md, junk). Each runs on what + * the header pass, the prober, the merges and the spots know by now. + */ +export const RULES: Rule[] = [ + { + name: 'dead', + says: 'every probed first article is gone from the server (taken down or expired)', + sql: `SELECT release_id FROM probe_candidates WHERE status IN ('done', 'missing', 'error') GROUP BY 1 HAVING bool_and(status = 'missing')`, + }, + { + name: 'fake', + says: 'named as a film, episode or music, but the content is a program', + // An archive whose entries are all executables, under a media name (its own or its entries'); or a file that is + // an executable by name or by its first bytes, under a media name. + sql: `SELECT i.release_id FROM release_inspections i JOIN releases r ON r.id = i.release_id + WHERE i.kind = 'archive' AND jsonb_array_length(i.result->'names') > 0 + AND NOT EXISTS (SELECT 1 FROM jsonb_array_elements_text(i.result->'names') n WHERE n !~* '${EXECUTABLE}') + AND (r.name ~* '${MEDIA}' OR (i.result->>'names') ~* '${MEDIA}') + UNION + SELECT i.release_id FROM release_inspections i JOIN releases r ON r.id = i.release_id + WHERE i.kind = 'lone' AND ((i.result->>'yencName') ~* '${EXECUTABLE}' OR i.result->>'type' = 'exe') + AND (r.name ~* '${MEDIA}' OR (i.result->>'yencName') ~* '${MEDIA}')`, + }, + { + name: 'repeated', + says: `the same file content in ${REPEATED_RELEASES}+ releases (test files, spam)`, + sql: `SELECT DISTINCT h.release_id FROM file_hashes h + JOIN (SELECT hash16k, size FROM file_hashes GROUP BY 1, 2 HAVING count(DISTINCT release_id) >= ${REPEATED_RELEASES}) x USING (hash16k, size)`, + }, + { + name: 'locked', + says: 'passworded, and no spot, NZB or title gives the password', + sql: `SELECT r.id FROM releases r WHERE r.passworded AND NOT EXISTS (SELECT 1 FROM release_passwords p WHERE p.release_id = r.id)`, + }, + { + name: 'unusable', + says: 'random names, no name inside, no par2 or inner-name merge, no spot, a day after settling', + sql: `SELECT r.id FROM releases r + WHERE r.settled_at < now() - interval '1 day' AND NOT r.readable AND r.inner_name IS NULL + AND NOT EXISTS (SELECT 1 FROM release_merges m WHERE m.into_id = r.id) + AND NOT EXISTS (SELECT 1 FROM release_spots s WHERE s.release_id = r.id) + AND NOT EXISTS (SELECT 1 FROM par2_sets p WHERE p.release_id = r.id)`, + }, +] + +export interface JudgeRound { + /** Per rule: releases it judges now, verdicts added, verdicts taken back. */ + rules: Record +} + +/** + * The junk judge (REBUILD.md, junk; voyager 1's second filter, D42): after the header pass's rules (garbage + * signatures, executables, …) a release is judged again once it's probed, by what its articles showed. Each + * round recomputes every rule: a release a rule matches gets a verdict (the first rule in order, over a later one's), one it no longer + * matches loses its verdict. Verdicts only hide (search leaves them out); nothing is dropped yet. + */ +export class ReleaseJudge { + private stopped = false + + stop() { + this.stopped = true + } + + async run() { + while (!this.stopped) { + const started = Date.now() + try { + const round = await this.round() + const parts = Object.entries(round.rules).map(([n, r]) => `${n} ${r.matched} (+${r.added} −${r.removed})`) + logger.info(`Judged: ${parts.join(', ')}.`) + } catch (error: any) { + logger.error(`Judge round failed: ${error.message}`) + } + for (let waited = Date.now() - started; waited < ROUND_MS && !this.stopped; waited += 1000) await new Promise((r) => setTimeout(r, 1000)) + } + } + + /** One round over every rule; with dry, only counts. */ + async round(dry = false): Promise { + const round: JudgeRound = { rules: {} } + // Verdicts of an older rule set are judged again from scratch. + if (!dry) await db.from('release_verdicts').where('rule_version', '<>', JUDGE_VERSION).delete() + const judged = new Set() + for (const rule of RULES) { + if (this.stopped) break + const { rows } = await db.rawQuery(rule.sql) + const ids = rows.map((r: Record) => String(Object.values(r)[0])).filter((id: string) => !judged.has(id)) + for (const id of ids) judged.add(id) + const stats = { matched: ids.length, added: 0, removed: 0 } + if (!dry) { + await db.transaction(async (trx) => { + // Dropped first: inside an outer transaction (the tests) ON COMMIT DROP comes only at its end. + await trx.rawQuery(`DROP TABLE IF EXISTS judged`) + await trx.rawQuery(`CREATE TEMP TABLE judged (release_id bigint PRIMARY KEY) ON COMMIT DROP`) + for (let i = 0; i < ids.length; i += 50_000) { + await trx.rawQuery(`INSERT INTO judged SELECT unnest(?::bigint[]) ON CONFLICT DO NOTHING`, [ids.slice(i, i + 50_000)]) + } + const removed = await trx.rawQuery( + `DELETE FROM release_verdicts v WHERE v.rule = ? AND NOT EXISTS (SELECT 1 FROM judged j WHERE j.release_id = v.release_id)`, + [rule.name] + ) + const added = await trx.rawQuery( + `INSERT INTO release_verdicts (release_id, rule, rule_version) + SELECT j.release_id, ?, ? FROM judged j JOIN releases r ON r.id = j.release_id + ON CONFLICT (release_id) DO UPDATE SET rule = excluded.rule, rule_version = excluded.rule_version + WHERE release_verdicts.rule <> excluded.rule`, + [rule.name, JUDGE_VERSION] + ) + stats.removed = removed.rowCount ?? 0 + stats.added = added.rowCount ?? 0 + }) + } + round.rules[rule.name] = stats + } + return round + } +} diff --git a/app/services/search_service.ts b/app/services/search_service.ts index 53c9131..45808fb 100644 --- a/app/services/search_service.ts +++ b/app/services/search_service.ts @@ -7,6 +7,8 @@ export type SortOrder = 'newest' | 'oldest' | 'largest' | 'files' | 'name' /** Voyager 1's search parameters, on the catalog's columns (REBUILD.md, part 9). */ export interface SearchParams { q?: string + /** Also releases the junk judge hid (release_verdicts). */ + judged?: boolean /** Only releases with a file whose name is readable (not an obfuscated token), or an inner name from the prober. */ readable?: boolean /** Only releases known to have all their files. */ @@ -80,6 +82,8 @@ function contains(word: string): string { class SearchService { async search(params: SearchParams) { const where: string[] = ['r.files > 0'] + // Judged junk (app/services/release_judge.ts) is left out unless asked for. + if (!params.judged) where.push('NOT EXISTS (SELECT 1 FROM release_verdicts v WHERE v.release_id = r.id)') const bindings: unknown[] = [] // Every word in the name or the inner name the prober found, through their trigram indexes; "The.Block.AU" and "the block au" alike. for (const word of (params.q ?? '').split(/[\s._-]+/).filter(Boolean).slice(0, 8)) { diff --git a/commands/releases_judge.ts b/commands/releases_judge.ts new file mode 100644 index 0000000..1f5a4b0 --- /dev/null +++ b/commands/releases_judge.ts @@ -0,0 +1,38 @@ +import { BaseCommand, flags } from '@adonisjs/core/ace' +import type { CommandOptions } from '@adonisjs/core/types/ace' +import { ReleaseJudge, RULES } from '#services/release_judge' + +/** The junk judge (REBUILD.md, junk): releases judged after their probes; verdicts hide them from search. */ +export default class ReleasesJudge extends BaseCommand { + static commandName = 'releases:judge' + static description = 'Judges releases by what their probes showed (dead, fake, repeated, locked, unusable), hourly' + static options: CommandOptions = { startApp: true, staysAlive: true } + + @flags.boolean({ description: 'One round, then exit' }) + declare once: boolean + + @flags.boolean({ description: 'With --once: count only, write no verdicts' }) + declare dry: boolean + + async run() { + const judge = new ReleaseJudge() + if (this.once) { + const round = await judge.round(this.dry) + for (const rule of RULES) { + const r = round.rules[rule.name] + if (r) this.logger.info(`${rule.name}: ${r.matched} (+${r.added} −${r.removed}) — ${rule.says}`) + } + await this.terminate() + return + } + let done: () => void + const finished = new Promise((resolve) => (done = resolve)) + this.app.terminating(async () => { + judge.stop() + await finished + }) + await judge.run() + this.logger.info('Judge stopped.') + done!() + } +} diff --git a/database/migrations/1770000003300_release_verdicts.ts b/database/migrations/1770000003300_release_verdicts.ts new file mode 100644 index 0000000..544bfa2 --- /dev/null +++ b/database/migrations/1770000003300_release_verdicts.ts @@ -0,0 +1,24 @@ +import { BaseSchema } from '@adonisjs/lucid/schema' + +/** + * The junk judge's verdicts (app/services/release_judge.ts): per release the rule that judged it, after its + * probes, and the rule set's version. Search leaves judged releases out. Hidden only for now (nothing dropped): + * a verdict whose rule no longer matches (a password found, a spot linked) is taken back by the next round. + */ +export default class extends BaseSchema { + async up() { + this.schema.raw(` + CREATE TABLE release_verdicts ( + release_id bigint PRIMARY KEY, + rule varchar(16) NOT NULL, + verdict varchar(8) NOT NULL DEFAULT 'hidden', + rule_version integer NOT NULL, + created_at timestamptz NOT NULL DEFAULT now() + )`) + this.schema.raw(`CREATE INDEX release_verdicts_rule_idx ON release_verdicts (rule)`) + } + + async down() { + this.schema.raw(`DROP TABLE IF EXISTS release_verdicts`) + } +} diff --git a/deploy/README.md b/deploy/README.md index eb44f00..586d2bd 100644 --- a/deploy/README.md +++ b/deploy/README.md @@ -20,6 +20,7 @@ homelab with the deploy key, whose forced command | `coordinator` | `usenet-rebuild` | prefetching (all NNTP connections), planning, the tail, merging, settling | | `prober` | `usenet-rebuild` | phase 2: reads the probe queue's articles over `PROBE_CONNECTIONS` (6) connections, stores what they show | | `merger` | `usenet-rebuild` | the par2 merge: releases whose files one par2 set describes become one (no NNTP) | +| `judge` | `usenet-rebuild` | the junk judge: releases judged hourly by what their probes showed (dead, fake, repeated, locked, unusable); verdicts in `release_verdicts` hide them from search (no NNTP) | | `spotter` | `usenet-rebuild` | Spotnet: free.pt's spots into `spots`, signatures checked, the passwords they name kept; 1 connection | | `spot-inspector` | `usenet-rebuild` | each spot's NZB: its files in `spot_files`, linked to the releases they match (`release_spots`), passwords in `release_passwords`; 1 connection | | `edge` | `usenet-rebuild-edge` (target `edge`) | the Rust worker, `--threads $EDGE_THREADS` (8) | diff --git a/deploy/compose.yaml b/deploy/compose.yaml index fd18d58..b01e561 100644 --- a/deploy/compose.yaml +++ b/deploy/compose.yaml @@ -122,6 +122,13 @@ services: command: ['node', 'ace', 'releases:merge'] volumes: [] + # The junk judge: releases judged hourly by what their probes showed (dead, fake, repeated, locked, unusable); + # verdicts hide them from search. No NNTP. + judge: + <<: *app + command: ['node', 'ace', 'releases:judge'] + volumes: [] + # Spotnet: the spotter reads free.pt's spots (verified, with the passwords # they name) into the spots table; the spot inspector reads each spot's NZB, # lists its files and links the releases they match. One connection each. diff --git a/deploy/deploy.sh b/deploy/deploy.sh index 6df14dd..587a799 100755 --- a/deploy/deploy.sh +++ b/deploy/deploy.sh @@ -45,10 +45,10 @@ docker compose run --rm --no-deps migrate # stream keeps the node API from exiting. echo "== recreating app services" docker compose stop edge -docker compose up -d --no-deps node-api web coordinator prober merger spotter spot-inspector +docker compose up -d --no-deps node-api web coordinator prober merger judge spotter spot-inspector docker compose up -d --no-deps edge echo "$SHA $(date -Is)" >> "$ROOT/deployed.log" # The images this replaced (321 MB each; the root filesystem is the NVMe, with a 10 GB floor). docker image prune -f --filter label=com.docker.compose.project=usenet-rebuild >/dev/null -docker compose ps --format '{{.Service}} {{.Status}}' node-api web coordinator prober merger spotter spot-inspector edge +docker compose ps --format '{{.Service}} {{.Status}}' node-api web coordinator prober merger judge spotter spot-inspector edge echo "== deployed $SHA" diff --git a/tests/functional/release_judge.spec.ts b/tests/functional/release_judge.spec.ts new file mode 100644 index 0000000..66dac23 --- /dev/null +++ b/tests/functional/release_judge.spec.ts @@ -0,0 +1,91 @@ +import { test } from '@japa/runner' +import db from '@adonisjs/lucid/services/db' +import testUtils from '@adonisjs/core/services/test_utils' +import { ReleaseJudge } from '#services/release_judge' +import SearchService from '#services/search_service' + +test.group('ReleaseJudge', (group) => { + group.each.setup(() => testUtils.db().withGlobalTransaction()) + + test('judges dead, fake, repeated, locked and unusable releases, and takes a verdict back', async ({ assert }) => { + let n = 0 + const settled = new Date(Date.now() - 3 * 86400_000) + const release = async (name: string, extra: Record = {}) => { + n++ + const key = `jj${String(n).padStart(6, '0')}`.padEnd(40, '0') + const [row] = await db + .table('releases') + .insert({ release_key: key, identity_key: key, name, source: 'stem', groups: '[]', group_id: 1, files: 3, posted_at: settled, readable: true, settled_at: settled, ...extra }) + .returning('id') + return Number(row.id) + } + let candidate = 9000 + const probe = async (releaseId: number, status: string) => { + candidate++ + await db.table('probe_candidates').insert({ id: candidate, release_id: releaseId, kind: 'lone', message_id: `<${candidate}@x>`, file: 'x', status }) + return candidate + } + const inspect = async (releaseId: number, kind: string, result: Record) => { + const id = await probe(releaseId, 'done') + await db.table('release_inspections').insert({ release_id: releaseId, candidate_id: id, kind, result: JSON.stringify(result) }) + } + + const dead = await release('Gone.Movie.2020.1080p') + await probe(dead, 'missing') + await probe(dead, 'missing') + const halfDead = await release('Half.Movie.2020.1080p') + await probe(halfDead, 'missing') + await probe(halfDead, 'done') + + const fakeArchive = await release('Das.A-Team.S02E23.Das.Spiel.ist.aus') + await inspect(fakeArchive, 'archive', { type: 'rar4', names: ['Das.A-Team.S02E23.Das.Spiel.ist.aus.exe'], entries: 1 }) + const fakeFile = await release('Some.Film.2021.1080p.WEB-DL') + await inspect(fakeFile, 'lone', { type: 'exe', yencName: 'Some.Film.2021.1080p.WEB-DL.mkv' }) + // A program under a program's name is software, not fake. + const software = await release('LimeWire 4.19.6') + await inspect(software, 'archive', { type: 'rar4', names: ['LimeWire 4.19.6.exe'], entries: 1 }) + + const hash = 'a'.repeat(32) + const repeatedIds: number[] = [] + for (let i = 0; i < 20; i++) { + const id = await release(`kantodo${i}`) + repeatedIds.push(id) + await db.table('file_hashes').insert({ candidate_id: 20000 + i, release_id: id, hash16k: hash, size: 1073741824 }) + } + + const locked = await release('Locked.Thing', { passworded: true }) + const unlocked = await release('Known.Password', { passworded: true }) + await db.table('release_passwords').insert({ release_id: unlocked, password: 'pw', source: 'spot-text' }) + + const unusable = await release('ab12CD34ef56GH78ij90', { readable: false }) + const fresh = await release('zz12CD34ef56GH78ij90', { readable: false, settled_at: new Date() }) + const spotted = await release('yy12CD34ef56GH78ij90', { readable: false }) + await db.table('release_spots').insert({ release_id: spotted, spot_id: 's@x', files: 1 }) + + const round = await new ReleaseJudge().round() + const verdicts = new Map((await db.from('release_verdicts').select('release_id', 'rule')).map((v) => [Number(v.release_id), v.rule])) + assert.equal(verdicts.get(dead), 'dead') + assert.isUndefined(verdicts.get(halfDead)) + assert.equal(verdicts.get(fakeArchive), 'fake') + assert.equal(verdicts.get(fakeFile), 'fake') + assert.isUndefined(verdicts.get(software)) + assert.deepEqual([...new Set(repeatedIds.map((id) => verdicts.get(id)))], ['repeated']) + assert.equal(verdicts.get(locked), 'locked') + assert.isUndefined(verdicts.get(unlocked)) + assert.equal(verdicts.get(unusable), 'unusable') + assert.isUndefined(verdicts.get(fresh)) + assert.isUndefined(verdicts.get(spotted)) + assert.equal(round.rules.repeated.added, 20) + + // Search leaves them out, unless asked. + const found = await SearchService.search({ q: 'Locked' }) + assert.equal(found.total, 0) + assert.equal((await SearchService.search({ q: 'Locked', judged: true })).total, 1) + + // A password found: the next round takes the verdict back. + await db.table('release_passwords').insert({ release_id: locked, password: 'found', source: 'nzb' }) + const again = await new ReleaseJudge().round() + assert.equal(again.rules.locked.removed, 1) + assert.isNull(await db.from('release_verdicts').where('release_id', locked).first()) + }) +})