From e135076fcd2fd821a45f59d0fe5ef866643c5899 Mon Sep 17 00:00:00 2001 From: Daan Meijer Date: Mon, 26 Oct 2026 13:41:30 +0100 Subject: [PATCH] The pending ladder in the prober: unnamed releases probed cheapest first, signatures learned, the named moved into the catalog pending_ladder.ts claims the oldest waiting pending releases' next target (par2, archive, lone, 7z-tail in that order) as at least PENDING_SHARE of each claim; a usable name (the largest named file of a par2 set or a listing, or a lone file's yEnc name, neither packaging nor random) hands the release to the namer, no name the next target or the end of the climb; climbs count on pending_signatures per (tool, group, poster) and (tool, group), and a signature with 100 climbs and 0.5% named or fewer is dropped: its waiting releases are discarded without a probe, the rules' sample excepted. pending_namer.ts reads a named release's articles again from the corpus or by XOVER and writes it to the catalog under the pending key (source weak-stem, the name as inner name) with its chunk and candidates. Migration 6400; config PENDING_*. Co-Authored-By: Claude Fable 5.1 --- REBUILD.md | 54 ++-- app/services/batch_store.ts | 11 +- app/services/corpus.ts | 16 +- app/services/pending_ladder.ts | 230 ++++++++++++++++++ app/services/pending_namer.ts | 222 +++++++++++++++++ app/services/prober.ts | 72 +++++- config/coordinator.ts | 8 + .../1770000006400_pending_ladder.ts | 49 ++++ lib/probe/index.ts | 2 +- lib/probe/inspect.ts | 62 ++++- start/env.ts | 3 + tests/functional/pending_ladder.spec.ts | 208 ++++++++++++++++ tests/functional/pending_namer.spec.ts | 143 +++++++++++ tests/unit/probe/inspect.spec.ts | 15 +- 14 files changed, 1045 insertions(+), 50 deletions(-) create mode 100644 app/services/pending_ladder.ts create mode 100644 app/services/pending_namer.ts create mode 100644 database/migrations/1770000006400_pending_ladder.ts create mode 100644 tests/functional/pending_ladder.spec.ts create mode 100644 tests/functional/pending_namer.spec.ts diff --git a/REBUILD.md b/REBUILD.md index e104bfe..45cdced 100644 --- a/REBUILD.md +++ b/REBUILD.md @@ -550,13 +550,17 @@ releases; the inspector matches them again), `nntp_shares`, `nodes`, `release_texts`. The category trigger reads the groups from `group_sets`; the judge's dirty trigger watches `poster_id` instead of `poster`. -- `pending_releases` (migration 6300, the HDD): the unnamed releases - outside the catalog (decision f): `release_key` (primary), `identity_key`, - `group_id`, `name` (the random stem), `poster` text, the posting span, - `first_article`, `last_article`, counts, `partial`, `finished`, `tool`, - `candidates` jsonb (the probe targets by effort), `status` (0 waiting, 1 - named, 2 discarded), `tries`, `created_at`, `probed_at`; indexed by - (group, tool, status, created_at) for the ladder. +- `pending_releases` (migrations 6300 and 6400, the HDD): the unnamed + releases outside the catalog (decision f): `release_key` (primary), + `identity_key`, `group_id`, `name` (the random stem), `poster` text, the + posting span, `first_article`, `last_article`, counts, `file_total`, + `partial`, `finished`, `tool`, `candidates` jsonb (the probe targets), + `inspections` jsonb (what each probe showed), `named`, `status` (0 + waiting, 1 named and in the catalog, 2 discarded, 3 exhausted, 4 named + awaiting the move, 9 taken), `tries`, `created_at`, `probed_at`; partial + indexes on the waiting (by `created_at`) and the named (by `probed_at`). +- `pending_signatures` (6400): per (tool, group, poster; poster '' for the + tool and group together) `probed`, `named`, `dropped`, `updated_at`. - `release_texts(release_id, kind smallint, text varchar(255))` (6200), primary key on all three, one trigram index (fastupdate): every string search matches against, appended by whoever learns one and never an @@ -689,12 +693,9 @@ until the window is done (LOG.md 2026-10-26 (111) has the first local numbers, (112) the homelab run). Next: the replay on homelab over the 2016 window (the NVMe bytes per -header for real), the pending ladder and the signature learner (decision -f, amended 2026-10-26: the prober working `pending_releases` cheapest -first, verdicts per tool and group, a named release moved into the -catalog with its segments read again), the ordered range leases with the -carried window in the edge-node and its snapshot (design 1 and 5), the -prober writing `release_texts` kind 2. Gone from the list: block +header for real), the ordered range leases with the carried window in +the edge-node and its snapshot (design 1 and 5), the prober writing +`release_texts` kind 2, the spot/NZB match on pending keys (decision f). Gone from the list: block leases, the planner and the block merger (design 1, amended). ## Decisions @@ -729,10 +730,29 @@ f. **Nothing enters the catalog without a usable name** (amended 11% of the segments and nothing anyone can find; the full history has 725 M random-named files (LOG.md 2026-10-09). Built 2026-10-26: the edge sends an unnamed release with an empty payload and its tool, the - batch store writes `pending_releases` (LOG.md (114)). Not built: the - ladder itself (the prober working `pending_releases` cheapest first, - the signature verdicts, the move into the catalog with the segments - read again), and the spot/NZB match on pending keys. + batch store writes `pending_releases` (LOG.md (114)); the ladder in the + prober (`app/services/pending_ladder.ts`, LOG.md (115)): a claim takes + the oldest waiting releases' next target (par2, archive, lone, 7z-tail + in that order), at least PENDING_SHARE (25%) of each claim next to the + catalog's own candidates, answered from `probe_cache` when the article + was read before; `usableName` (lib/probe/inspect.ts) takes the largest + named file of a par2 set or a listing, or a lone file's yEnc name, that + is neither packaging nor random; a name sets status 4, no name the next + target or status 3 (exhausted); each finished climb counts on + `pending_signatures` for (tool, group, poster) and (tool, group), and a + signature with PENDING_MIN_PROBES (100) climbs and at most + PENDING_MAX_NAMED (0.5%) named is dropped: every five minutes the + waiting releases of dropped signatures are discarded (status 2), the + rules' `samplePercent` of them left to climb. The namer + (`app/services/pending_namer.ts`, in the prober's process): a named + release's article span read again from the corpus (the stored range's + chunks) or by XOVER on a connection of its own, through the TypeScript + indexer, written as a release under the pending key with `source` + weak-stem, the name as `inner_name`, its chunk and its candidates (the + prober answers those from `probe_cache`: inspections, par2 sets, hashes, + the password flag). Not built: the spot/NZB match on pending keys (step + 0 of the ladder beyond signatures), and a signature by poster alone + across groups. g. The account's NNTP connections shared by priority (`nntp_shares`): the coordinator (headers) before the prober, each between its minimum and maximum, 5 kept for the spotter, the spot inspector, the plain diff --git a/app/services/batch_store.ts b/app/services/batch_store.ts index 28c94aa..9efb2b2 100644 --- a/app/services/batch_store.ts +++ b/app/services/batch_store.ts @@ -147,7 +147,7 @@ class BatchStore { } /** The range's block; a range without one (no planner yet) is a block of its own, leased as the range is. */ - private async blockOf(trx: TransactionClientContract, range: any): Promise { + async blockOf(trx: TransactionClientContract, range: any): Promise { if (range.block_id) return Number(range.block_id) const [block] = await trx .table('blocks') @@ -204,21 +204,22 @@ class BatchStore { bytes: r.bytes, partial: r.partial === true, finished: r.finished !== false, + file_total: r.fileTotal, tool: r.tool ?? (probes[0] ? uploadTool(probes[0].messageId) : 'unknown'), candidates: probes, } }) const { rows: done } = await trx.rawQuery( `INSERT INTO pending_releases (release_key, identity_key, group_id, name, poster, posted_at, last_posted_at, first_article, last_article, - files, segments, bytes, partial, finished, tool, candidates) + files, segments, bytes, partial, finished, file_total, tool, candidates) SELECT decode(v.release_key, 'hex'), decode(v.identity_key, 'hex'), ?, v.name, v.poster, to_timestamp(v.first / 1000.0), to_timestamp(v.last / 1000.0), - v.first_article, v.last_article, v.files, v.segments, v.bytes, v.partial, v.finished, v.tool, v.candidates + v.first_article, v.last_article, v.files, v.segments, v.bytes, v.partial, v.finished, v.file_total, v.tool, v.candidates FROM jsonb_to_recordset(?::jsonb) AS v(release_key text, identity_key text, name text, poster text, first bigint, last bigint, first_article bigint, last_article bigint, - files int, segments bigint, bytes bigint, partial boolean, finished boolean, tool text, candidates jsonb) + files int, segments bigint, bytes bigint, partial boolean, finished boolean, file_total int, tool text, candidates jsonb) ON CONFLICT (release_key) DO UPDATE SET files = pending_releases.files + excluded.files, segments = pending_releases.segments + excluded.segments, bytes = pending_releases.bytes + excluded.bytes, last_posted_at = greatest(pending_releases.last_posted_at, excluded.last_posted_at), last_article = greatest(pending_releases.last_article, excluded.last_article), - partial = excluded.partial, finished = excluded.finished, + partial = excluded.partial, finished = excluded.finished, file_total = coalesce(excluded.file_total, pending_releases.file_total), candidates = CASE WHEN jsonb_array_length(pending_releases.candidates) = 0 THEN excluded.candidates ELSE pending_releases.candidates END WHERE NOT pending_releases.finished RETURNING release_key`, diff --git a/app/services/corpus.ts b/app/services/corpus.ts index 80212ae..1295d31 100644 --- a/app/services/corpus.ts +++ b/app/services/corpus.ts @@ -42,13 +42,13 @@ class Corpus { return coordinatorConfig.corpus.dir } - private rangeDir(range: CorpusRange) { - return join(this.dir!, range.group, `${range.start}-${range.end}`) + private rangeDir(range: CorpusRange, dir = this.dir!) { + return join(dir, range.group, `${range.start}-${range.end}`) } - private path(range: CorpusRange, span: ChunkSpan) { + private path(range: CorpusRange, span: ChunkSpan, dir = this.dir!) { const index = Math.floor((span.from - range.start) / coordinatorConfig.chunkArticles) - return join(this.rangeDir(range), `${String(index).padStart(4, '0')}-${span.from}-${span.to}.gz`) + return join(this.rangeDir(range, dir), `${String(index).padStart(4, '0')}-${span.from}-${span.to}.gz`) } /** Whether every chunk of the range is stored (its range.json). */ @@ -69,10 +69,10 @@ class Corpus { ) } - /** A stored chunk, or null. */ - async read(range: CorpusRange, span: ChunkSpan): Promise { - if (!this.dir) return null - return readFile(this.path(range, span)).catch(() => null) + /** A stored chunk, or null (`dir`: another corpus than CORPUS_DIR's, the namer's tests). */ + async read(range: CorpusRange, span: ChunkSpan, dir = this.dir): Promise { + if (!dir) return null + return readFile(this.path(range, span, dir)).catch(() => null) } /** diff --git a/app/services/pending_ladder.ts b/app/services/pending_ladder.ts new file mode 100644 index 0000000..fa7176b --- /dev/null +++ b/app/services/pending_ladder.ts @@ -0,0 +1,230 @@ +import db from '@adonisjs/lucid/services/db' +import type { TransactionClientContract } from '@adonisjs/lucid/types/database' +import coordinatorConfig from '#config/coordinator' +import { explored } from '#lib/rules/index' +import { usableName, type Inspection } from '#lib/probe/index' +import RuleSets from '#services/rule_sets' + +/** pending_releases.status (migration 6400). */ +export const PENDING = { + waiting: 0, + named: 1, + discarded: 2, + exhausted: 3, + naming: 4, + taken: 9, +} as const + +/** The probe targets of a pending release in the order of effort: a par2 names the set's files outright, an archive's start its entries, a lone file only its yEnc name, a 7z's tail is the costliest find. */ +export const EFFORT = ['par2', 'archive', 'lone', '7z-tail'] as const + +export interface PendingTarget { + kind: string + messageId: string + file: string +} + +/** One probe the ladder hands the prober: the next target of a pending release. */ +export interface PendingProbe { + /** pending_releases.release_key as hex. */ + releaseKey: string + kind: string + messageId: string + file: string + tool: string + groupId: number + poster: string + /** Targets probed before this one. */ + tries: number + targets: number +} + +export type PendingOutcome = { + probe: PendingProbe + status: 'done' | 'missing' | 'error' | 'waiting' + inspection?: Inspection +} + +export function byEffort(targets: PendingTarget[]): PendingTarget[] { + const rank = (kind: string) => { + const i = (EFFORT as readonly string[]).indexOf(kind) + return i < 0 ? EFFORT.length : i + } + return [...targets].sort((a, b) => rank(a.kind) - rank(b.kind)) +} + +/** + * The pending ladder (REBUILD.md decision f): pending_releases probed from least to most effort, a usable name + * handing the release to the namer (app/services/pending_namer.ts), no name after the last target exhausting it; + * every finished climb counted on its signature (tool, group, poster; and the tool and group together), and a + * signature with enough probes and next to no names dropped: its waiting releases are discarded without a probe, + * a sample of them (the rules' samplePercent) still climbing so the rate keeps being measured. + */ +class PendingLadder { + /** A previous prober's claims back in the queue. */ + async release(): Promise { + const { rowCount } = await db.rawQuery( + `UPDATE pending_releases SET status = ? WHERE status = ?`, + [PENDING.waiting, PENDING.taken] + ) + return rowCount ?? 0 + } + + /** The next target of up to `n` waiting releases with a target left, oldest first, claimed. */ + async claim(n: number): Promise { + if (n <= 0) return [] + const { rows } = await db.rawQuery( + `UPDATE pending_releases p SET status = ? FROM ( + SELECT release_key FROM pending_releases + WHERE status = ? AND finished AND tries < jsonb_array_length(candidates) + ORDER BY created_at LIMIT ? FOR UPDATE SKIP LOCKED) c + WHERE p.release_key = c.release_key + RETURNING encode(p.release_key, 'hex') AS release_key, p.candidates, p.tries, p.tool, p.group_id, coalesce(p.poster, '') AS poster`, + [PENDING.taken, PENDING.waiting, n] + ) + return rows.map((r: any) => { + const targets = byEffort(r.candidates as PendingTarget[]) + const tries = Number(r.tries) + const target = targets[Math.min(tries, targets.length - 1)] + return { + releaseKey: r.release_key, + kind: target.kind, + messageId: target.messageId, + file: target.file, + tool: r.tool, + groupId: Number(r.group_id), + poster: r.poster, + tries, + targets: targets.length, + } + }) + } + + /** + * What the probes showed: a usable name moves the release to the namer (status 4), else the next target, or the + * end of the ladder (status 3); the finished climbs counted on their signatures. + */ + async record(outcomes: PendingOutcome[], trx: TransactionClientContract | typeof db = db) { + if (outcomes.length === 0) return + const rows = outcomes.map((o) => { + const p = o.probe + if (o.status === 'waiting') { + return { + release_key: p.releaseKey, + status: PENDING.waiting, + tries: p.tries, + named: null, + inspection: null, + final: null, + } + } + const name = o.status === 'done' && o.inspection ? usableName(o.inspection, p.kind) : null + const tries = p.tries + 1 + const inspection = { + kind: p.kind, + status: o.status, + result: o.inspection ? { ...o.inspection, par2: undefined } : null, + } + if (name) + return { + release_key: p.releaseKey, + status: PENDING.naming, + tries: 0, + named: name, + inspection, + final: 'named', + } + if (tries >= p.targets) + return { + release_key: p.releaseKey, + status: PENDING.exhausted, + tries, + named: null, + inspection, + final: 'exhausted', + } + return { + release_key: p.releaseKey, + status: PENDING.waiting, + tries, + named: null, + inspection, + final: null, + } + }) + await trx.rawQuery( + `UPDATE pending_releases p SET status = v.status, tries = v.tries, named = coalesce(v.named, p.named), + inspections = CASE WHEN v.inspection IS NULL THEN p.inspections ELSE p.inspections || v.inspection END, + probed_at = CASE WHEN v.inspection IS NULL THEN p.probed_at ELSE now() END + FROM jsonb_to_recordset(?::jsonb) AS v(release_key text, status int, tries int, named text, inspection jsonb) + WHERE p.release_key = decode(v.release_key, 'hex')`, + [JSON.stringify(rows.map(({ final: _final, ...row }) => row))] + ) + // The signatures: the poster's own and the tool's and group's together. + const counts = new Map< + string, + { tool: string; group_id: number; poster: string; probed: number; named: number } + >() + for (const [i, o] of outcomes.entries()) { + const final = rows[i].final + if (!final) continue + for (const poster of [o.probe.poster, '']) { + const key = `${o.probe.tool}\t${o.probe.groupId}\t${poster}` + const entry = counts.get(key) ?? { + tool: o.probe.tool, + group_id: o.probe.groupId, + poster, + probed: 0, + named: 0, + } + entry.probed++ + if (final === 'named') entry.named++ + counts.set(key, entry) + } + } + if (counts.size === 0) return + const { pendingMinProbes, pendingMaxNamed } = coordinatorConfig.probe + await trx.rawQuery( + `INSERT INTO pending_signatures (tool, group_id, poster, probed, named) + SELECT v.tool, v.group_id, v.poster, v.probed, v.named FROM jsonb_to_recordset(?::jsonb) AS v(tool text, group_id int, poster text, probed int, named int) + ON CONFLICT (tool, group_id, poster) DO UPDATE SET probed = pending_signatures.probed + excluded.probed, named = pending_signatures.named + excluded.named, updated_at = now()`, + [JSON.stringify([...counts.values()])] + ) + await trx.rawQuery( + `UPDATE pending_signatures s SET dropped = (s.probed >= ?::int AND s.named::float <= s.probed * ?::float) + FROM jsonb_to_recordset(?::jsonb) AS v(tool text, group_id int, poster text) + WHERE s.tool = v.tool AND s.group_id = v.group_id AND s.poster = v.poster`, + [ + pendingMinProbes, + pendingMaxNamed, + JSON.stringify( + [...counts.values()].map(({ tool, group_id, poster }) => ({ tool, group_id, poster })) + ), + ] + ) + } + + /** Waiting releases of a dropped signature discarded without a probe, the explored sample excepted; how many. */ + async discard(limit = 10_000): Promise { + const { samplePercent } = await RuleSets.current() + const { rows } = await db.rawQuery( + `SELECT encode(p.release_key, 'hex') AS release_key FROM pending_releases p + WHERE p.status = ? AND EXISTS ( + SELECT 1 FROM pending_signatures s + WHERE s.dropped AND s.tool = p.tool AND s.group_id = p.group_id AND (s.poster = '' OR s.poster = coalesce(p.poster, ''))) + LIMIT ?`, + [PENDING.waiting, limit] + ) + const keys = rows + .map((r: any) => r.release_key as string) + .filter((key: string) => !explored(key, samplePercent)) + if (keys.length === 0) return 0 + const { rowCount } = await db.rawQuery( + `UPDATE pending_releases SET status = ? WHERE status = ? AND release_key = ANY(SELECT decode(k, 'hex') FROM unnest(?::text[]) AS k)`, + [PENDING.discarded, PENDING.waiting, keys] + ) + return rowCount ?? 0 + } +} + +export default new PendingLadder() diff --git a/app/services/pending_namer.ts b/app/services/pending_namer.ts new file mode 100644 index 0000000..8840404 --- /dev/null +++ b/app/services/pending_namer.ts @@ -0,0 +1,222 @@ +import db from '@adonisjs/lucid/services/db' +import logger from '@adonisjs/core/services/logger' +import { setTimeout as sleep } from 'node:timers/promises' +import { promisify } from 'node:util' +import { unzip } from 'node:zlib' +import env from '#start/env' +import coordinatorConfig from '#config/coordinator' +import { XoverClient } from '#lib/nntp/index' +import { RangeIndexer, type ReleaseChunk } from '#lib/edge/index' +import { identityKey } from '#lib/release/keys' +import BatchStore from '#services/batch_store' +import Corpus from '#services/corpus' +import RuleSets from '#services/rule_sets' +import { PENDING } from '#services/pending_ladder' + +const inflate = promisify(unzip) + +interface NamedRow { + release_key: Buffer + identity_key: Buffer + group_id: number + named: string + first_article: number + last_article: number + file_total: number | null + tries: number +} + +/** The namer gives a release up after this many failed moves (its articles gone from the server and the corpus). */ +const MAX_MOVES = 3 + +/** + * The move into the catalog (REBUILD.md decision f): a pending release a probe named (status 4) gets its segments + * read again, from the corpus when its range is stored, else from the server (XOVER over its article span, one + * connection of its own, opened for it and closed after), through the TypeScript indexer, and is written as a + * release with the name the probe found (source weak-stem, the name as inner name), its chunk and its probe + * candidates, which the prober answers from probe_cache. Runs inside the prober. + */ +export class PendingNamer { + private client: XoverClient | null = null + private usedAt = 0 + + constructor(private corpusDir: string | undefined = coordinatorConfig.corpus.dir) {} + + async run(stopped: () => boolean) { + while (!stopped()) { + let rows: NamedRow[] = [] + try { + rows = await this.take(20) + for (const row of rows) { + if (stopped()) break + try { + await this.move(row) + } catch (error: any) { + logger.warn( + `Naming ${row.release_key.toString('hex')} "${row.named}" failed: ${error.message}` + ) + await db.rawQuery( + `UPDATE pending_releases SET tries = tries + 1, probed_at = now() WHERE release_key = ?`, + [row.release_key] + ) + } + } + } catch (error: any) { + logger.error(`Naming: ${error.message}`) + } + if (this.client && Date.now() - this.usedAt > coordinatorConfig.nntpIdleCloseMs) { + this.client.close() + this.client = null + } + if (rows.length === 0) + for (let waited = 0; waited < 30_000 && !stopped(); waited += 1000) await sleep(1000) + } + this.client?.close() + this.client = null + } + + /** Named releases awaiting their move, the longest waiting first; those failed MAX_MOVES times stay as they are. */ + async take(n: number): Promise { + const { rows } = await db.rawQuery( + `SELECT release_key, identity_key, group_id, named, first_article, last_article, file_total, tries + FROM pending_releases WHERE status = ? AND tries < ? ORDER BY probed_at LIMIT ?`, + [PENDING.naming, MAX_MOVES, n] + ) + return rows.map((r: any) => ({ + ...r, + group_id: Number(r.group_id), + first_article: Number(r.first_article), + last_article: Number(r.last_article), + tries: Number(r.tries), + })) + } + + /** One named release into the catalog. */ + async move(row: NamedRow): Promise { + const group = await db.from('groups').where('id', row.group_id).firstOrFail() + const range = await db + .from('ranges') + .where('group_id', row.group_id) + .where('start_id', '<=', row.first_article) + .where('end_id', '>=', row.first_article) + .orderBy('id') + .first() + if (!range) throw new Error(`no range of ${group.name} holds article ${row.first_article}`) + const text = await this.overviewText(group.name, row) + const rules = await RuleSets.current() + const ix = new RangeIndexer({ + group: group.name, + start: row.first_article, + end: row.last_article, + rules: { ...rules, samplePercent: 0 }, + idleChunks: 0, + edgeArticles: 0, + }) + ix.addText(text, row.last_article + 1) + ix.finish() + const batch = ix.take() + const found = batch.releases.find((r) => identityKey(r.key).equals(row.identity_key)) + if (!found) + throw new Error( + `its articles ${row.first_article}–${row.last_article} of ${group.name} no longer show it` + ) + // The catalog row under the pending key (the identity, the group, the first article the edge saw). + const chunk: ReleaseChunk = { + ...found, + name: row.named, + source: 'weak-stem', + readable: true, + finished: true, + edge: false, + firstArticle: row.first_article, + fileTotal: row.file_total ?? found.fileTotal, + tool: undefined, + } + const candidates = batch.candidates + .filter((c) => c.releaseKey === found.key) + .map((c) => ({ ...c, upload: found.upload })) + await db.transaction(async (trx) => { + const blockId = await BatchStore.blockOf(trx, range) + await BatchStore.writeReleases( + trx, + { groupId: row.group_id, blockId }, + row.first_article, + [chunk], + candidates + ) + await trx.rawQuery( + `UPDATE releases SET inner_name = ?, inspected_at = now() WHERE release_key = ? AND inner_name IS NULL`, + [row.named.slice(0, 255), row.release_key] + ) + await trx.rawQuery( + `UPDATE pending_releases SET status = ?, probed_at = now() WHERE release_key = ?`, + [PENDING.named, row.release_key] + ) + }) + } + + /** The overview lines of the release's article span: the stored corpus chunks when its range is stored, else XOVER. */ + private async overviewText(groupName: string, row: NamedRow): Promise { + if (this.corpusDir) { + const stored = await db + .from('ranges') + .where('group_id', row.group_id) + .whereNotNull('stored_at') + .where('start_id', '<=', row.last_article) + .where('end_id', '>=', row.first_article) + .orderBy('start_id') + const parts: string[] = [] + let covered = true + for (const r of stored) { + const range = { group: groupName, start: Number(r.start_id), end: Number(r.end_id) } + const size = coordinatorConfig.chunkArticles + for ( + let from = + range.start + + Math.floor((Math.max(row.first_article, range.start) - range.start) / size) * size; + from <= Math.min(row.last_article, range.end); + from += size + ) { + const raw = await Corpus.read( + range, + { from, to: Math.min(from + size - 1, range.end) }, + this.corpusDir + ) + if (!raw) { + covered = false + break + } + parts.push((await inflate(raw)).toString('latin1')) + } + if (!covered) break + } + if ( + covered && + parts.length > 0 && + stored.length > 0 && + Number(stored[0].start_id) <= row.first_article && + Number(stored[stored.length - 1].end_id) >= row.last_article + ) { + return parts.join('') + } + } + this.client ??= await XoverClient.connect({ + host: env.get('NNTP_HOST')!, + port: env.get('NNTP_PORT', 563), + user: env.get('NNTP_USER'), + password: env.get('NNTP_PASSWORD'), + secure: env.get('NNTP_SECURE', true), + compress: true, + timeoutMs: coordinatorConfig.nntpTimeoutMs, + }) + this.usedAt = Date.now() + try { + await this.client.group(groupName) + return (await this.client.xoverRaw(row.first_article, row.last_article)).toString('latin1') + } catch (error) { + this.client.close() + this.client = null + throw error + } + } +} diff --git a/app/services/prober.ts b/app/services/prober.ts index 245c278..5a33d93 100644 --- a/app/services/prober.ts +++ b/app/services/prober.ts @@ -9,12 +9,16 @@ import { inspect, innerName, PROBE_VERSION, type Inspection } from '#lib/probe/i import { NntpShare } from '#services/nntp_shares' import { DiskSpace } from '#services/disk_space' import { decidePar2Completeness } from '#services/par2_completeness' +import PendingLadder, { type PendingProbe } from '#services/pending_ladder' +import { PendingNamer } from '#services/pending_namer' interface Candidate { id: number releaseId: number kind: string messageId: string + /** A pending release's target (the pending ladder), not a catalog candidate: id and releaseId are 0. */ + pending?: PendingProbe } type Outcome = { candidate: Candidate; status: 'done' | 'missing' | 'error' | 'waiting'; inspection?: Inspection; cached?: boolean } @@ -34,6 +38,8 @@ const TIERS: { kind: string; unreadable: boolean }[] = [ { kind: '7z-tail', unreadable: false }, ] const FLUSH_MS = 2000 +/** Between sweeps that discard the waiting releases of dropped signatures. */ +const SWEEP_MS = 5 * 60_000 /** * JSON for a jsonb parameter, without NUL characters: jsonb can't hold \u0000 (a name read from an article had one), @@ -72,13 +78,19 @@ export class Prober { /** Per minute: what was read, and where the lanes' time went (ms summed over lanes; flushes ms on their own). */ private static freshStats() { - return { articles: 0, bytes: 0, done: 0, missing: 0, error: 0, cached: 0, waitMs: 0, idleMs: 0, fetchMs: 0, inspectMs: 0, flushMs: 0, claimMs: 0, since: Date.now() } + return { articles: 0, bytes: 0, done: 0, missing: 0, error: 0, cached: 0, pending: 0, discarded: 0, waitMs: 0, idleMs: 0, fetchMs: 0, inspectMs: 0, flushMs: 0, claimMs: 0, since: Date.now() } } + /** The pending ladder's namer: named pending releases into the catalog, alongside the lanes. */ + private namer = new PendingNamer() + private sweptAt = 0 async run() { // A previous run's claims: one prober at a time. const { rowCount } = await db.rawQuery(`UPDATE probe_candidates SET status = 'waiting' WHERE status = 'taken'`) if (rowCount) logger.info(`Probing: ${rowCount} candidates from a previous run back in the queue.`) + const released = await PendingLadder.release() + if (released) logger.info(`Probing: ${released} pending releases from a previous run back on the ladder.`) + const namer = this.namer.run(() => this.stopped).catch((error) => logger.error(`Naming failed: ${error.message}`)) logger.info(`Probing with ${this.share.min}–${this.share.max} connections (its share of the account's).`) const flusher = (async () => { while (!this.stopped || this.outcomes.length > 0) { @@ -108,6 +120,7 @@ export class Prober { for (let waited = 0; waited < coordinatorConfig.nntpShares.everyMs && !this.stopped; waited += 1000) await sleep(1000) } await Promise.all(this.lanes) + await namer await this.share.leave().catch(() => {}) // What was claimed and not probed goes back. for (const candidate of this.queue.splice(0)) this.outcomes.push({ candidate, status: 'waiting' }) @@ -227,6 +240,37 @@ export class Prober { private async claimTiers() { // The queue empties, the lanes go dry and the share shrinks to its minimum until there's room again. if (await this.space.short()) return + await this.claimCatalog() + await this.claimPending() + } + + /** The pending ladder's share of a claim: at least pendingShare of the batch, and whatever the catalog's tiers left. */ + private async claimPending() { + if (Date.now() - this.sweptAt > SWEEP_MS) { + this.sweptAt = Date.now() + this.stats.discarded += await PendingLadder.discard() + } + const want = Math.max(Math.floor(this.batch * coordinatorConfig.probe.pendingShare), this.batch - this.queue.length) + const probes = await PendingLadder.claim(want) + if (probes.length === 0) return + const { rows: cached } = await db.rawQuery(`SELECT message_id, status, result FROM probe_cache WHERE message_id = ANY(?) AND version = ?`, [ + probes.map((p) => p.messageId), + PROBE_VERSION, + ]) + const known = new Map(cached.map((c: any) => [c.message_id, c])) + for (const probe of probes) { + const candidate: Candidate = { id: 0, releaseId: 0, kind: probe.kind, messageId: probe.messageId, pending: probe } + const hit: any = known.get(probe.messageId) + if (hit) { + this.stats.cached++ + this.outcomes.push({ candidate, status: hit.status, inspection: hit.result ?? undefined, cached: true }) + } else { + this.queue.push(candidate) + } + } + } + + private async claimCatalog() { for (const { kind, unreadable } of TIERS) { const want = this.batch - this.queue.length if (want <= 0) return @@ -277,18 +321,25 @@ export class Prober { /** The outcomes so far: candidates' status, inspections, and the releases' passworded and inner name. */ private async flush() { - const outcomes = this.outcomes.splice(0) - if (outcomes.length === 0) return - for (const o of outcomes) if (o.status !== 'waiting') this.stats[o.status]++ + const taken = this.outcomes.splice(0) + if (taken.length === 0) return + for (const o of taken) if (o.status !== 'waiting') this.stats[o.status]++ + // The ladder's outcomes apart from the catalog's: the next target, a name for the namer, or the end of the climb. + const pending = taken.filter((o) => o.candidate.pending) + const outcomes = taken.filter((o) => !o.candidate.pending) + this.stats.pending += pending.filter((o) => o.status !== 'waiting').length await db.transaction(async (trx) => { - await trx.rawQuery( - `UPDATE probe_candidates pc SET status = v.status FROM jsonb_to_recordset(?::jsonb) AS v(id bigint, status text) WHERE pc.id = v.id`, - [jsonb(outcomes.map((o) => ({ id: o.candidate.id, status: o.status })))] - ) + if (outcomes.length > 0) { + await trx.rawQuery( + `UPDATE probe_candidates pc SET status = v.status FROM jsonb_to_recordset(?::jsonb) AS v(id bigint, status text) WHERE pc.id = v.id`, + [jsonb(outcomes.map((o) => ({ id: o.candidate.id, status: o.status })))] + ) + } + await PendingLadder.record(pending.map((o) => ({ probe: o.candidate.pending!, status: o.status, inspection: o.inspection })), trx) // Final answers read from the news server, kept by message-id for the next catalog (a reset). // One per message-id: the same article is a candidate of more than one release, and ON CONFLICT can't take a // row twice in one statement. - const fresh = [...new Map(outcomes.filter((o) => !o.cached && o.status !== 'waiting').map((o) => [o.candidate.messageId, o])).values()] + const fresh = [...new Map(taken.filter((o) => !o.cached && o.status !== 'waiting').map((o) => [o.candidate.messageId, o])).values()] if (fresh.length > 0) { await trx.rawQuery( `INSERT INTO probe_cache (message_id, status, result, version) @@ -365,7 +416,8 @@ export class Prober { const { articles, bytes, done, missing, error } = this.stats logger.info( `Probed ${articles} articles in ${seconds.toFixed(0)} s: ${(articles / seconds).toFixed(1)}/s, ${((bytes * 8) / seconds / 1e6).toFixed(0)} Mbit/s; ` + - `${done} inspected, ${missing} missing, ${error} not yEnc; ${this.stats.cached} from the probe cache. ` + + `${done} inspected, ${missing} missing, ${error} not yEnc; ${this.stats.cached} from the probe cache; ` + + `the pending ladder ${this.stats.pending} probed, ${this.stats.discarded} discarded by signature. ` + `Lanes: fetching ${(this.stats.fetchMs / 1000).toFixed(0)} s, waiting for candidates ${(this.stats.waitMs / 1000).toFixed(0)} s, ` + `idle ${(this.stats.idleMs / 1000).toFixed(0)} s, inspecting ${(this.stats.inspectMs / 1000).toFixed(1)} s; ` + `claims ${(this.stats.claimMs / 1000).toFixed(1)} s, flushes ${(this.stats.flushMs / 1000).toFixed(1)} s.` diff --git a/config/coordinator.ts b/config/coordinator.ts index b4f25af..8541402 100644 --- a/config/coordinator.ts +++ b/config/coordinator.ts @@ -103,6 +103,14 @@ export default { batch: 400, /** BODY commands in flight per connection (pipelined; XoverClient.bodies). */ pipeline: env.get('PROBE_PIPELINE', 8), + /** + * The pending ladder (REBUILD.md decision f; app/services/pending_ladder.ts): the share of each claim taken + * from pending_releases at least (the catalog's own candidates come first, the rest of the claim is theirs). + */ + pendingShare: env.get('PENDING_SHARE', 0.25), + /** A signature (tool, group, poster) with this many climbs and at most pendingMaxNamed of them named is dropped. */ + pendingMinProbes: env.get('PENDING_MIN_PROBES', 100), + pendingMaxNamed: env.get('PENDING_MAX_NAMED', 0.005), }, /** * The catalog's tablespace may grow to this many GB (app/services/catalog_space.ts); above it only the tail and diff --git a/database/migrations/1770000006400_pending_ladder.ts b/database/migrations/1770000006400_pending_ladder.ts new file mode 100644 index 0000000..b6cbce4 --- /dev/null +++ b/database/migrations/1770000006400_pending_ladder.ts @@ -0,0 +1,49 @@ +import { BaseSchema } from '@adonisjs/lucid/schema' + +/** + * The pending ladder (REBUILD.md decision f, 2026-10-26): what the prober needs to work pending_releases cheapest + * first and to learn which signatures never name anything. + * - pending_releases: `file_total` (the counter's, for the catalog row once named), `inspections` (what each probe + * showed, by kind), `named` (the usable name a probe found; status 4 until the namer moved it into the catalog). + * Status: 0 waiting, 1 named and in the catalog, 2 discarded (its signature's verdict), 3 exhausted (every + * target probed, no name), 4 named, awaiting the move, 9 taken by a running prober. + * - pending_signatures: per (tool, group, poster) how many pending releases were probed to the end and how many + * got a name; `dropped` once probed ≥ PENDING_MIN_PROBES and named ≤ PENDING_MAX_NAMED of them (voyager 2's + * garbage learner, LOG.md 2026-10-08 (5)). The poster '' row is the tool's and group's together. + */ +export default class extends BaseSchema { + async up() { + this.schema.raw(`ALTER TABLE pending_releases ADD COLUMN IF NOT EXISTS file_total integer`) + this.schema.raw( + `ALTER TABLE pending_releases ADD COLUMN IF NOT EXISTS inspections jsonb NOT NULL DEFAULT '[]'` + ) + this.schema.raw(`ALTER TABLE pending_releases ADD COLUMN IF NOT EXISTS named varchar(255)`) + // The ladder's claim: the oldest waiting ones; the namer's: the named ones. + this.schema.raw( + `CREATE INDEX IF NOT EXISTS pending_releases_waiting_idx ON pending_releases (created_at) WHERE status = 0` + ) + this.schema.raw( + `CREATE INDEX IF NOT EXISTS pending_releases_naming_idx ON pending_releases (probed_at) WHERE status = 4` + ) + this.schema.raw(` + CREATE TABLE pending_signatures ( + tool varchar(16) NOT NULL, + group_id integer NOT NULL, + poster text NOT NULL DEFAULT '', + probed integer NOT NULL DEFAULT 0, + named integer NOT NULL DEFAULT 0, + dropped boolean NOT NULL DEFAULT false, + updated_at timestamptz NOT NULL DEFAULT now(), + PRIMARY KEY (tool, group_id, poster) + )`) + } + + async down() { + this.schema.raw(`DROP TABLE IF EXISTS pending_signatures`) + this.schema.raw(`DROP INDEX IF EXISTS pending_releases_naming_idx`) + this.schema.raw(`DROP INDEX IF EXISTS pending_releases_waiting_idx`) + this.schema.raw(`ALTER TABLE pending_releases DROP COLUMN IF EXISTS named`) + this.schema.raw(`ALTER TABLE pending_releases DROP COLUMN IF EXISTS inspections`) + this.schema.raw(`ALTER TABLE pending_releases DROP COLUMN IF EXISTS file_total`) + } +} diff --git a/lib/probe/index.ts b/lib/probe/index.ts index 51cc1f1..1658a59 100644 --- a/lib/probe/index.ts +++ b/lib/probe/index.ts @@ -1 +1 @@ -export { inspect, innerName, MAX_NAMES, PROBE_VERSION, type Inspection } from './inspect.js' +export { inspect, innerName, usableName, MAX_NAMES, PROBE_VERSION, type Inspection } from './inspect.js' diff --git a/lib/probe/inspect.ts b/lib/probe/inspect.ts index ba747b6..281cff2 100644 --- a/lib/probe/inspect.ts +++ b/lib/probe/inspect.ts @@ -5,6 +5,7 @@ import { parseRarHeaders } from '#lib/rar/index' import { parseSevenZipTail } from '#lib/sevenzip/index' import { parseZipStart } from '#lib/zip/index' import { sniffContentType, type ContentType } from '#lib/content/index' +import { fileStem, looksRandom } from '#lib/release/index' /** What one probed article showed (REBUILD.md, part 6; stored per release and kind). */ export interface Inspection { @@ -25,7 +26,10 @@ export interface Inspection { */ hash16k?: string /** A par2 body's recovery set: every file it describes (not kept in release_inspections; par2_files). */ - par2?: { setId: string; files: { fileId: string; hash16k: string; length: number; name: string }[] } + par2?: { + setId: string + files: { fileId: string; hash16k: string; length: number; name: string }[] + } } const HASH_BYTES = 16 * 1024 @@ -63,9 +67,16 @@ function largestFirst(items: { name: string; size: number }[]): string[] { */ export function inspect(body: Buffer, kind?: string): Inspection { const segment = decodeBody(body) - const result: Inspection = { yencName: segment.meta.begin.name.slice(0, MAX_NAME_LENGTH), size: segment.meta.begin.size, type: sniffContentType(segment.data) } + const result: Inspection = { + yencName: segment.meta.begin.name.slice(0, MAX_NAME_LENGTH), + size: segment.meta.begin.size, + type: sniffContentType(segment.data), + } // The file's first 16 KiB, or all of a smaller file. - if (segment.offset === 0 && (segment.data.length >= HASH_BYTES || segment.data.length === result.size)) { + if ( + segment.offset === 0 && + (segment.data.length >= HASH_BYTES || segment.data.length === result.size) + ) { result.hash16k = createHash('md5').update(segment.data.subarray(0, HASH_BYTES)).digest('hex') } if (isPar2(segment.data)) { @@ -76,7 +87,12 @@ export function inspect(body: Buffer, kind?: string): Inspection { result.names = largestFirst(set.files.map((f) => ({ name: f.name, size: f.length }))) result.par2 = { setId: set.recoverySetId, - files: set.files.map((f) => ({ fileId: f.fileId, hash16k: f.md5_16k, length: f.length, name: f.name.slice(0, MAX_NAME_LENGTH) })), + files: set.files.map((f) => ({ + fileId: f.fileId, + hash16k: f.md5_16k, + length: f.length, + name: f.name.slice(0, MAX_NAME_LENGTH), + })), } } return result @@ -99,7 +115,10 @@ export function inspect(body: Buffer, kind?: string): Inspection { } } // 7z keeps its listing at the end: the last segment of the last volume (a 7z-tail), or a small archive whole. - if ((kind === '7z-tail' || result.type === '7z') && segment.offset + segment.data.length === result.size) { + if ( + (kind === '7z-tail' || result.type === '7z') && + segment.offset + segment.data.length === result.size + ) { const listing = parseSevenZipTail(segment.data) if (listing) { result.entries = listing.files.length @@ -110,16 +129,43 @@ export function inspect(body: Buffer, kind?: string): Inspection { return result } +/** + * The usable name one probe of an unnamed release found (the pending ladder, REBUILD.md decision f): the largest + * named file of a par2 set or an archive's listing, or a lone file's yEnc name, when it isn't packaging and isn't + * random itself (an obfuscated upload's par2 names its random volumes; its yEnc name is mostly the random name + * again). Null: this probe named nothing, the next target is tried. + */ +export function usableName(inspection: Inspection, kind: string): string | null { + const usable = (n: string | undefined) => { + if (!n) return null + const name = n.replace(/^.*[\\/]/, '').slice(0, 255) + return name && !PACKAGING.test(name) && !looksRandom(fileStem(name)) ? name : null + } + const named = (inspection.names ?? []).map(usable).find((n) => n) + if (named) return named + return kind === 'lone' || !inspection.names?.length ? usable(inspection.yencName) : null +} + /** * A release's best inner name from its inspections (by probe kind): the * largest named entry of an archive, else the largest file of a par2 set * that isn't packaging, else a lone file's yEnc name. Null when none says * more than the release's own name. */ -export function innerName(inspections: { kind: string; result: Inspection }[], releaseName = ''): string | null { +export function innerName( + inspections: { kind: string; result: Inspection }[], + releaseName = '' +): string | null { const of = (kind: string) => inspections.filter((i) => i.kind === kind).map((i) => i.result) - const content = (kind: string) => of(kind).flatMap((i) => i.names ?? []).find((n) => !PACKAGING.test(n)) - const pick = content('archive') ?? content('7z-tail') ?? content('par2') ?? of('lone').find((i) => i.yencName && !PACKAGING.test(i.yencName))?.yencName + const content = (kind: string) => + of(kind) + .flatMap((i) => i.names ?? []) + .find((n) => !PACKAGING.test(n)) + const pick = + content('archive') ?? + content('7z-tail') ?? + content('par2') ?? + of('lone').find((i) => i.yencName && !PACKAGING.test(i.yencName))?.yencName const name = pick?.replace(/^.*[\\/]/, '').slice(0, 255) // The release's own name with an extension says nothing new (a random-named file's yEnc name is mostly its random name). return name && !name.toLowerCase().startsWith(releaseName.toLowerCase() || '\0') ? name : null diff --git a/start/env.ts b/start/env.ts index 7c74004..19e9cbf 100644 --- a/start/env.ts +++ b/start/env.ts @@ -74,6 +74,9 @@ export default await Env.create(new URL('../', import.meta.url), { SPOT_STANDALONE_AFTER_HOURS: Env.schema.number.optional(), CORPUS_AHEAD: Env.schema.number.optional(), PROBE_PIPELINE: Env.schema.number.optional(), + PENDING_SHARE: Env.schema.number.optional(), + PENDING_MIN_PROBES: Env.schema.number.optional(), + PENDING_MAX_NAMED: Env.schema.number.optional(), SPOOL_AHEAD_PER_LEASE: Env.schema.number.optional(), SPOOL_MIN_AHEAD: Env.schema.number.optional(), SPOOL_MAX_GB: Env.schema.number.optional(), diff --git a/tests/functional/pending_ladder.spec.ts b/tests/functional/pending_ladder.spec.ts new file mode 100644 index 0000000..a275ed1 --- /dev/null +++ b/tests/functional/pending_ladder.spec.ts @@ -0,0 +1,208 @@ +import { test } from '@japa/runner' +import db from '@adonisjs/lucid/services/db' +import testUtils from '@adonisjs/core/services/test_utils' +import PendingLadder, { PENDING, byEffort, type PendingProbe } from '#services/pending_ladder' +import { identityKey, releaseKey } from '#lib/release/keys' + +let groupId = 0 +let n = 0 + +async function pending( + name: string, + candidates: { kind: string; messageId: string; file: string }[], + extra: Record = {} +) { + n++ + const key = releaseKey(name, groupId, n) + await db.table('pending_releases').insert({ + release_key: key, + identity_key: identityKey(name), + group_id: groupId, + name, + poster: 'Bob ', + first_article: n, + last_article: n, + files: 1, + segments: 10, + bytes: 1000, + tool: 'other', + candidates: JSON.stringify(candidates), + ...extra, + }) + return key.toString('hex') +} + +const inspection = (yencName: string, names?: string[]) => ({ + yencName, + size: 1000, + type: null, + names, + entries: names?.length, +}) + +test.group('Pending ladder', (group) => { + group.each.setup(async () => { + const rollback = await testUtils.db().withGlobalTransaction() + groupId = Number( + ( + await db + .table('groups') + .insert({ name: 'alt.binaries.pending', priority: 1 }) + .returning('id') + )[0].id + ) + n = 0 + return rollback + }) + + test('targets are climbed by effort: a par2 before an archive before a lone file', ({ + assert, + }) => { + assert.deepEqual( + byEffort([ + { kind: 'lone', messageId: 'l', file: 'l' }, + { kind: '7z-tail', messageId: 't', file: 't' }, + { kind: 'archive', messageId: 'a', file: 'a' }, + { kind: 'par2', messageId: 'p', file: 'p' }, + ]).map((t) => t.kind), + ['par2', 'archive', 'lone', '7z-tail'] + ) + }) + + test('a claim takes the oldest waiting releases, their cheapest target first; outcomes move them up or to the end', async ({ + assert, + }) => { + const twoTargets = await pending('a1b2c3d4e5f6a1b2c3d4e5f6a1b2c3d4', [ + { kind: 'archive', messageId: '', file: 'x.rar' }, + { kind: 'par2', messageId: '', file: 'x.par2' }, + ]) + const lone = await pending('f6e5d4c3b2a1f6e5d4c3b2a1f6e5d4c3', [ + { kind: 'lone', messageId: '', file: 'f6e5d4c3b2a1f6e5d4c3b2a1f6e5d4c3' }, + ]) + const none = await pending('00000000000000000000000000000001', []) + const provisional = await pending( + '00000000000000000000000000000002', + [{ kind: 'lone', messageId: '', file: 'v' }], + { finished: false } + ) + + const probes = await PendingLadder.claim(10) + assert.deepEqual( + probes.map((p) => [p.releaseKey, p.kind, p.tries, p.targets]), + [ + [twoTargets, 'par2', 0, 2], + [lone, 'lone', 0, 1], + ] + ) + const statusOf = async (key: string) => + (await db + .from('pending_releases') + .where('release_key', Buffer.from(key, 'hex')) + .firstOrFail()) as any + assert.equal((await statusOf(twoTargets)).status, PENDING.taken) + assert.equal((await statusOf(none)).status, PENDING.waiting) + assert.equal((await statusOf(provisional)).status, PENDING.waiting) + + // The par2 names only random volumes: the next target. The lone file's yEnc name is its random name: exhausted. + await PendingLadder.record([ + { + probe: probes[0], + status: 'done', + inspection: inspection('x.par2', ['a1b2c3d4e5f6a1b2c3d4e5f6a1b2c3d4.part01.rar']), + }, + { + probe: probes[1], + status: 'done', + inspection: inspection('f6e5d4c3b2a1f6e5d4c3b2a1f6e5d4c3'), + }, + ]) + const first = await statusOf(twoTargets) + assert.include(first, { status: PENDING.waiting, tries: 1, named: null }) + assert.lengthOf(first.inspections, 1) + assert.include(await statusOf(lone), { status: PENDING.exhausted, tries: 1 }) + + // The archive's listing names it: for the namer. + const [second] = await PendingLadder.claim(10) + assert.deepEqual([second.releaseKey, second.kind, second.tries], [twoTargets, 'archive', 1]) + await PendingLadder.record([ + { + probe: second, + status: 'done', + inspection: inspection('x.rar', ['Some.Movie.2016.1080p.mkv']), + }, + ]) + assert.include(await statusOf(twoTargets), { + status: PENDING.naming, + named: 'Some.Movie.2016.1080p.mkv', + tries: 0, + }) + + // The signature counted both ways: the poster's own and the tool's in the group. + const signatures = await db.from('pending_signatures').orderBy('poster') + assert.deepEqual( + signatures.map((s) => [s.tool, s.poster, s.probed, s.named, s.dropped]), + [ + ['other', '', 2, 1, false], + ['other', 'Bob ', 2, 1, false], + ] + ) + }) + + test('a signature that names nothing is dropped, and its waiting releases discarded without a probe', async ({ + assert, + }) => { + // 100 climbs of the tool in the group, none named. + await db + .table('pending_signatures') + .insert({ tool: 'other', group_id: groupId, poster: '', probed: 99, named: 0 }) + const last = await pending('00000000000000000000000000000009', [ + { kind: 'lone', messageId: '', file: 'z' }, + ]) + const [probe] = await PendingLadder.claim(1) + assert.equal(probe.releaseKey, last) + await PendingLadder.record([{ probe, status: 'missing' }]) + const signature = await db + .from('pending_signatures') + .where({ tool: 'other', group_id: groupId, poster: '' }) + .firstOrFail() + assert.include(signature, { probed: 100, named: 0, dropped: true }) + + const waiting = await pending('00000000000000000000000000000010', [ + { kind: 'lone', messageId: '', file: 'w' }, + ]) + const other = await pending( + '00000000000000000000000000000011', + [{ kind: 'lone', messageId: '', file: 'o' }], + { tool: 'nyuu' } + ) + // samplePercent 0 in the test rules: no explored sample. + await db.table('rule_sets').insert({ + version: 99, + rules: JSON.stringify({ samplePercent: 0, droppedSignatures: [] }), + note: 'test', + }) + assert.equal(await PendingLadder.discard(), 1) + const statusOf = async (key: string) => + ( + await db + .from('pending_releases') + .where('release_key', Buffer.from(key, 'hex')) + .firstOrFail() + ).status + assert.equal(await statusOf(waiting), PENDING.discarded) + assert.equal(await statusOf(other), PENDING.waiting) + }) + + test('a probe given back goes to the queue again at the same step', async ({ assert }) => { + const key = await pending('00000000000000000000000000000020', [ + { kind: 'lone', messageId: '', file: 'g' }, + ]) + const [probe] = await PendingLadder.claim(1) + await PendingLadder.record([{ probe: probe as PendingProbe, status: 'waiting' }]) + assert.include( + await db.from('pending_releases').where('release_key', Buffer.from(key, 'hex')).firstOrFail(), + { status: PENDING.waiting, tries: 0 } + ) + assert.lengthOf(await db.from('pending_signatures'), 0) + }) +}) diff --git a/tests/functional/pending_namer.spec.ts b/tests/functional/pending_namer.spec.ts new file mode 100644 index 0000000..ea347c4 --- /dev/null +++ b/tests/functional/pending_namer.spec.ts @@ -0,0 +1,143 @@ +import { test } from '@japa/runner' +import db from '@adonisjs/lucid/services/db' +import testUtils from '@adonisjs/core/services/test_utils' +import { mkdtemp, mkdir, writeFile, rm } from 'node:fs/promises' +import { tmpdir } from 'node:os' +import { join } from 'node:path' +import { deflateSync } from 'node:zlib' +import { PendingNamer } from '#services/pending_namer' +import { PENDING } from '#services/pending_ladder' +import { RangeIndexer } from '#lib/edge/index' +import { DEFAULT_RULES } from '#lib/rules/index' +import { decodeChunk } from '#lib/chunks/index' +import { identityKey, releaseKey } from '#lib/release/keys' + +const GROUP = 'alt.binaries.namer' + +/** One overview line as XOVER gives it (the corpus stores the reply's text). */ +function line(n: number, subject: string, minutes: number) { + const date = new Date(Date.UTC(2016, 0, 4, 10, minutes)).toUTCString() + return `${n}\t${subject}\tp \t${date}\t\t\t1000\t10\tXref: news.example ${GROUP}:${n}` +} + +test.group('Pending namer', (group) => { + let dir = '' + group.each.setup(async () => { + const rollback = await testUtils.db().withGlobalTransaction() + dir = await mkdtemp(join(tmpdir(), 'namer-')) + return rollback + }) + group.each.teardown(() => rm(dir, { recursive: true, force: true })) + + test('a named pending release is read again from the corpus and written to the catalog with its chunk and candidates', async ({ + assert, + }) => { + const groupId = Number( + (await db.table('groups').insert({ name: GROUP, priority: 1 }).returning('id'))[0].id + ) + // The range stored in the corpus: one chunk with a random-named upload (4 parts and its par2) among others. + const lines = [ + line(5, 'Some.Show [1/1] - "Some.Show.mkv" yEnc (1/1)', 0), + ...[1, 2, 3, 4].map((i) => + line(9 + i, `[1/2] - "q3B9OrpKiCxUnzgtUGnp.part01.rar" yEnc (${i}/4)`, i) + ), + line(14, '[2/2] - "q3B9OrpKiCxUnzgtUGnp.par2" yEnc (1/1)', 5), + line(20, 'Other.Thing [1/1] - "Other.Thing.mkv" yEnc (1/1)', 6), + ] + await mkdir(join(dir, GROUP, '1-200000'), { recursive: true }) + await writeFile( + join(dir, GROUP, '1-200000', '0000-1-100000.gz'), + deflateSync(Buffer.from(lines.join('\r\n') + '\r\n', 'latin1')) + ) + await db.table('ranges').insert({ + group_id: groupId, + start_id: 1, + end_id: 200_000, + next_id: 200_001, + status: 'done', + stored_at: new Date(), + capture: true, + }) + + // The key the edge gave it: the same identity rules, run here over the same lines. + const ix = new RangeIndexer({ + group: GROUP, + start: 1, + end: 200_000, + rules: { ...DEFAULT_RULES, droppedSignatures: [], priorities: { [GROUP]: 1 } }, + }) + ix.addText(lines.join('\r\n'), 200_001) + ix.finish() + const edge = ix.take().releases.find((r) => !r.readable)! + assert.equal(edge.firstArticle, 10) + const key = releaseKey(edge.key, groupId, 10) + await db.table('pending_releases').insert({ + release_key: key, + identity_key: identityKey(edge.key), + group_id: groupId, + name: 'q3B9OrpKiCxUnzgtUGnp', + poster: 'p ', + first_article: 10, + last_article: 14, + files: 2, + segments: 5, + bytes: 5000, + file_total: 2, + tool: 'other', + candidates: JSON.stringify([ + { kind: 'archive', messageId: 'u10@test', file: 'q3B9OrpKiCxUnzgtUGnp.part01.rar' }, + ]), + status: PENDING.naming, + named: 'Some.Movie.2016.1080p.BluRay.x264-GRP.mkv', + }) + + const namer = new PendingNamer(dir) + const [row] = await namer.take(5) + assert.equal(row.named, 'Some.Movie.2016.1080p.BluRay.x264-GRP.mkv') + await namer.move(row) + + const release = await db.from('releases').where('release_key', key).firstOrFail() + assert.include(release, { + name: 'Some.Movie.2016.1080p.BluRay.x264-GRP.mkv', + inner_name: 'Some.Movie.2016.1080p.BluRay.x264-GRP.mkv', + source: 7, + readable: true, + files: 2, + segments: '5', + file_total: 2, + complete: true, + first_article: '10', + last_article: '14', + }) + assert.isNotNull(release.settled_at) + const chunk = await db.from('release_chunks').where('release_id', release.id).firstOrFail() + assert.deepEqual( + decodeChunk(chunk.payload) + .map((f) => [f.name, f.segments.length]) + .sort(), + [ + ['q3B9OrpKiCxUnzgtUGnp.par2', 1], + ['q3B9OrpKiCxUnzgtUGnp.part01.rar', 4], + ] + ) + assert.deepEqual( + (await db.from('probe_candidates').where('release_id', release.id).orderBy('kind')).map( + (c) => [c.kind, c.ready, c.unreadable] + ), + [ + ['archive', true, false], + ['par2', true, false], + ] + ) + assert.deepEqual( + await db.from('release_texts').where('release_id', release.id).select('kind', 'text'), + [{ kind: 1, text: 'Some.Movie.2016.1080p.BluRay.x264-GRP.mkv' }] + ) + assert.equal( + (await db.from('pending_releases').where('release_key', key).firstOrFail()).status, + PENDING.named + ) + // Nothing else of the chunk got in. + assert.equal(Number((await db.from('releases').count('* as n').first())!.n), 1) + }) +}) diff --git a/tests/unit/probe/inspect.spec.ts b/tests/unit/probe/inspect.spec.ts index b6888b2..9548e45 100644 --- a/tests/unit/probe/inspect.spec.ts +++ b/tests/unit/probe/inspect.spec.ts @@ -4,7 +4,7 @@ import { readFileSync } from 'node:fs' import { crc32 } from 'node:zlib' import yencode from 'yencode' import { crc32Hex } from '#lib/yenc/index' -import { inspect, innerName, MAX_NAMES } from '#lib/probe/index' +import { inspect, innerName, MAX_NAMES, usableName } from '#lib/probe/index' /** A single-part yEnc article of `data`. */ function article(data: Buffer, name: string) { @@ -76,6 +76,19 @@ function rar5File(name: string, size: number, opts: { packed: number; headerFlag return rar5Block(2, opts.headerFlags ?? 0, fields, { extra, data: opts.packed }) } +test.group('probe / usable name (the pending ladder)', () => { + test('a par2 or listing that names a real file names the release; random volumes and packaging do not', ({ assert }) => { + const base = { size: 1000, type: null } + assert.equal(usableName({ ...base, yencName: 'x.par2', names: ['q3B9OrpKiCxUnzgtUGnp.part01.rar', 'Some.Movie.2016.mkv'] }, 'par2'), 'Some.Movie.2016.mkv') + assert.isNull(usableName({ ...base, yencName: 'q3B9OrpKiCxUnzgtUGnp.par2', names: ['q3B9OrpKiCxUnzgtUGnp.part01.rar'] }, 'par2')) + assert.equal(usableName({ ...base, yencName: 'x.rar', names: ['Folder/Real.Name.S01E01.mkv'] }, 'archive'), 'Real.Name.S01E01.mkv') + assert.isNull(usableName({ ...base, yencName: 'x.rar', names: [], passworded: true }, 'archive')) + assert.equal(usableName({ ...base, yencName: 'brothalovers.20.02.16.hazel.moore.mp4' }, 'lone'), 'brothalovers.20.02.16.hazel.moore.mp4') + assert.isNull(usableName({ ...base, yencName: 'c3453cdd2a3645a8f3cf3d22319abe63' }, 'lone')) + assert.isNull(usableName({ ...base, yencName: 'c3453cdd2a3645a8f3cf3d22319abe63.part01.rar' }, 'lone')) + }) +}) + test.group('probe / inspect', () => { test('a par2 index: its files, largest first', ({ assert }) => { const setId = Buffer.alloc(16, 3)