From 44c0c73b0584d6acf3efb9904965771f53c951bb Mon Sep 17 00:00:00 2001 From: Daan Meijer Date: Mon, 2 Nov 2026 12:22:32 +0100 Subject: [PATCH] Pending ladder by class, twins by content, dropped classes at the edge MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The prober names 6% of the unnamed releases it finishes and the pending queue (44M) grows eleven times faster than it probes. A sample of 44,530 finished pending rows shows the headers predict a name: size bucket, name shape, completeness, tool and group separate 0% classes from 95% ones, while the learner keyed on tool and group alone (LOG.md (130)). - lib/release/classes.ts, names::pending_class, SQL pending_class(): a release's class (size × shape × partial, 1..70), the three in step. - Migration 6800: pending_releases.klass and target_bytes (the first target's article; candidates carry bytes now), pending_classes with each class's climbs, names, waiting and verdict, the claim's index. - The ladder claims the classes with the best record first (a prior for the unexplored, at most a quarter of a claim from one), the cheapest target first within one, then the oldest of the rest; records classes like signatures and drops them at the same thresholds; discards their waiting releases. Claimed rows sorted in code: an UPDATE returns none. - The rule set's droppedClasses: the edge keeps unnamed releases of a dropped class out of the pending store (counted as garbage), the explored sample excepted. - hash_twins.ts: a probed file whose first 16 KiB and size match a named catalog release's names a pending release the probe didn't, and sets the inner name of a catalog release without one. - pending:classes classes the rows from before and seeds the record from the finished ones; rules:classes publishes the dropped classes; releases:readable also marks the rows' candidates unreadable. Co-Authored-By: Claude Fable 5.1 --- REBUILD.md | 15 +- app/services/batch_store.ts | 39 ++- app/services/hash_twins.ts | 27 ++ app/services/pending_ladder.ts | 259 +++++++++++++++--- app/services/prober.ts | 13 +- commands/pending_classes.ts | 97 +++++++ commands/releases_readable.ts | 11 +- commands/rules_classes.ts | 54 ++++ .../1770000006800_pending_classes.ts | 57 ++++ edge/src/indexer.rs | 17 +- edge/src/lease.rs | 3 + edge/src/names.rs | 50 ++++ edge/src/wire.rs | 3 +- lib/edge/range_indexer.ts | 26 +- lib/release/classes.ts | 55 ++++ lib/release/index.ts | 1 + lib/rules/rule_set.ts | 8 + tests/functional/pending_classes.spec.ts | 28 ++ tests/functional/pending_ladder.spec.ts | 197 ++++++++++++- tests/unit/release/classes.spec.ts | 22 ++ 20 files changed, 927 insertions(+), 55 deletions(-) create mode 100644 app/services/hash_twins.ts create mode 100644 commands/pending_classes.ts create mode 100644 commands/rules_classes.ts create mode 100644 database/migrations/1770000006800_pending_classes.ts create mode 100644 lib/release/classes.ts create mode 100644 tests/functional/pending_classes.spec.ts create mode 100644 tests/unit/release/classes.spec.ts diff --git a/REBUILD.md b/REBUILD.md index 223b966..7eab73d 100644 --- a/REBUILD.md +++ b/REBUILD.md @@ -897,7 +897,20 @@ f. **Nothing enters the catalog without a usable name** (amended yields none counts against its signature, and a signature with ≥100 probes and ≤0.5% named (voyager 2's garbage learner, LOG.md 2026-10-08 (5)) is **discarded** from then on, a sample of it still - probed to notice a change. The reason, in the 2016 replay: one poster's + probed to notice a change. **By class** (2026-11-02, LOG.md (130)): a + pending release's class is its tool, group, size bucket, name shape and + partial flag (`lib/release/classes.ts`, `names::pending_class`, the SQL + `pending_class()`; `pending_releases.klass`), and `pending_classes` keeps + each class's climbs and names; the claim takes the classes with the best + record first (`classScore`, a prior for the unexplored, at most a quarter + of a claim from one class), the cheapest target article first within one + (`target_bytes`), and the oldest of the rest; a class with the learner's + thresholds is dropped like a signature, and the rule set's + `droppedClasses` keep its releases out of the pending store at the edge, + the explored sample excepted. A probed file whose first 16 KiB and size + match a named catalog release's (`file_hashes`) takes that name without + another fetch (`app/services/hash_twins.ts`), on the ladder and for + catalog releases without one. The reason, in the 2016 replay: one poster's hash-named pieces ("Bob", 11 aliases, 3.2 M releases in five small groups) were 76% of the rows and three quarters of the NVMe bytes for 11% of the segments and nothing anyone can find; the full history has diff --git a/app/services/batch_store.ts b/app/services/batch_store.ts index 9efb2b2..70d8b42 100644 --- a/app/services/batch_store.ts +++ b/app/services/batch_store.ts @@ -9,6 +9,7 @@ import { SIDECAR } from '#services/sidecar_folder' import { identityKey, releaseKey } from '#lib/release/keys' import { sourceCode } from '#lib/release/source' import { uploadTool } from '#lib/rules/index' +import { byEffort } from '#services/pending_ladder' import { TEXT_KINDS, TEXT_MAX, textsToStore } from '#lib/release/text_kind' /** @@ -179,11 +180,14 @@ class BatchStore { candidates: ProbeCandidate[] ): Promise { if (releases.length === 0) return 0 - const targets = new Map() + const targets = new Map< + string, + { kind: string; messageId: string; file: string; bytes?: number }[] + >() for (const c of candidates) { const key = slot(c.releaseKey, c.upload ?? 0) const list = targets.get(key) ?? [] - list.push({ kind: c.kind, messageId: c.messageId, file: c.file }) + list.push({ kind: c.kind, messageId: c.messageId, file: c.file, bytes: c.bytes }) targets.set(key, list) } let written = 0 @@ -207,25 +211,46 @@ class BatchStore { file_total: r.fileTotal, tool: r.tool ?? (probes[0] ? uploadTool(probes[0].messageId) : 'unknown'), candidates: probes, + // The first target's article (by effort): the cheapest probe first within a class (LOG.md (130)). + target_bytes: byEffort(probes)[0]?.bytes ?? null, } }) 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, file_total, tool, candidates) + files, segments, bytes, partial, finished, file_total, tool, candidates, klass, target_bytes) 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.file_total, 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, + pending_class(v.name, v.bytes, v.partial), v.target_bytes 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, file_total int, tool text, candidates jsonb) + files int, segments bigint, bytes bigint, partial boolean, finished boolean, file_total int, tool text, candidates jsonb, target_bytes bigint) 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, 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 + candidates = CASE WHEN jsonb_array_length(pending_releases.candidates) = 0 THEN excluded.candidates ELSE pending_releases.candidates END, + klass = pending_class(pending_releases.name, pending_releases.bytes + excluded.bytes, excluded.partial), + target_bytes = coalesce(pending_releases.target_bytes, excluded.target_bytes) WHERE NOT pending_releases.finished - RETURNING release_key`, + RETURNING release_key, tool, klass, (xmax = 0) AS inserted`, [groupId, JSON.stringify(rows)] ) written += done.length + // Each class seen gets its row in the ladder's record, so the claim explores it (pending_classes). + const classes = new Map() + for (const d of done) { + const key = `${d.tool}\t${d.klass}` + const entry = classes.get(key) ?? { tool: d.tool, klass: Number(d.klass), waiting: 0 } + if (d.inserted) entry.waiting++ + classes.set(key, entry) + } + if (classes.size > 0) { + await trx.rawQuery( + `INSERT INTO pending_classes (tool, group_id, klass, waiting) + SELECT v.tool, ?, v.klass, v.waiting FROM jsonb_to_recordset(?::jsonb) AS v(tool text, klass int, waiting int) + ON CONFLICT (tool, group_id, klass) DO UPDATE SET waiting = pending_classes.waiting + excluded.waiting`, + [groupId, JSON.stringify([...classes.values()])] + ) + } } return written } diff --git a/app/services/hash_twins.ts b/app/services/hash_twins.ts new file mode 100644 index 0000000..0254703 --- /dev/null +++ b/app/services/hash_twins.ts @@ -0,0 +1,27 @@ +import db from '@adonisjs/lucid/services/db' + +/** + * Twins by content (LOG.md (130)): a probed file's first 16 KiB and size (file_hashes, what the par2 merge and the + * repeated verdict match) in a catalog release that has a name, readable or found inside. The same film posted + * under scrambles by other tools gets that name without another fetch. Releases with a junk verdict don't count. + */ +export async function twinNames( + pairs: { hash16k: string; size: number }[], + trx: { rawQuery: typeof db.rawQuery } = db +): Promise> { + const out = new Map() + if (pairs.length === 0) return out + const { rows } = await trx.rawQuery( + `SELECT DISTINCT ON (h.hash16k, h.size) h.hash16k, h.size, coalesce(r.inner_name, r.name) AS name + FROM jsonb_to_recordset(?::jsonb) AS v(hash16k text, size bigint) + JOIN file_hashes h ON h.hash16k = v.hash16k AND h.size = v.size + JOIN releases r ON r.id = h.release_id AND (r.readable OR r.inner_name IS NOT NULL) + WHERE NOT EXISTS (SELECT 1 FROM release_verdicts x WHERE x.release_id = r.id) + ORDER BY h.hash16k, h.size, r.readable DESC, r.id`, + [JSON.stringify(pairs)] + ) + for (const r of rows) out.set(`${r.hash16k}\t${r.size}`, r.name) + return out +} + +export const twinKey = (hash16k: string, size: number) => `${hash16k}\t${size}` diff --git a/app/services/pending_ladder.ts b/app/services/pending_ladder.ts index fa7176b..5328dfe 100644 --- a/app/services/pending_ladder.ts +++ b/app/services/pending_ladder.ts @@ -4,6 +4,7 @@ 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' +import { twinKey, twinNames } from '#services/hash_twins' /** pending_releases.status (migration 6400). */ export const PENDING = { @@ -22,6 +23,8 @@ export interface PendingTarget { kind: string messageId: string file: string + /** The target article's bytes (candidates from 2026-11-02 on). */ + bytes?: number } /** One probe the ladder hands the prober: the next target of a pending release. */ @@ -33,6 +36,8 @@ export interface PendingProbe { file: string tool: string groupId: number + /** Its class (lib/release/classes.ts; 0 for a row from before the classes). */ + klass: number poster: string /** Targets probed before this one. */ tries: number @@ -53,12 +58,69 @@ export function byEffort(targets: PendingTarget[]): PendingTarget[] { return [...targets].sort((a, b) => rank(a.kind) - rank(b.kind)) } +/** A class's record (pending_classes), for the claim's order. */ +interface ClassRow { + tool: string + groupId: number + klass: number + probed: number + named: number + waiting: number +} + +/** The record is read again this often. */ +const CLASSES_MS = 60_000 +/** A class found short of a claim's share is left alone this long. */ +const EMPTY_MS = 5 * 60_000 +/** The prior rate of an unexplored class and its weight in probes: the ladder's overall rate on 2026-11-02 (LOG.md (130)). */ +const PRIOR_RATE = 0.15 +const PRIOR_WEIGHT = 10 +/** One class takes at most this share of a claim: the rest explores the classes after it. */ +const CLASS_SHARE = 0.25 + +/** The expected naming rate of a class: its record shrunk towards the prior. */ +export function classScore(c: { probed: number; named: number }): number { + return (c.named + PRIOR_RATE * PRIOR_WEIGHT) / (c.probed + PRIOR_WEIGHT) +} + +/** An UPDATE returns its rows in no order: the claim's (the cheapest target, then the oldest) restored. */ +function claimed(rows: any[]): any[] { + const bytes = (r: any) => + r.target_bytes === null ? Number.POSITIVE_INFINITY : Number(r.target_bytes) + return [...rows].sort( + (a, b) => + bytes(a) - bytes(b) || + new Date(a.created_at).getTime() - new Date(b.created_at).getTime() || + Number(a.first_article) - Number(b.first_article) + ) +} + +function toProbe(r: any): PendingProbe { + 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), + klass: Number(r.klass ?? 0), + poster: r.poster, + tries, + targets: targets.length, + } +} + /** * 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. + * every finished climb counted on its signature (tool, group, poster; and the tool and group together) and on its + * class (tool, group, size bucket × name shape × partial; lib/release/classes.ts, LOG.md (130)), and a signature + * or class 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. The claim takes the + * classes with the best record first (classScore), the cheapest target first within one. */ class PendingLadder { /** A previous prober's claims back in the queue. */ @@ -70,34 +132,84 @@ class PendingLadder { return rowCount ?? 0 } - /** The next target of up to `n` waiting releases with a target left, oldest first, claimed. */ + /** The ladder's record per class, cached CLASSES_MS between claims, best first; classes found empty, until when. */ + private classes: ClassRow[] = [] + private classesAt = 0 + private empty = new Map() + + /** The record read again on the next claim (the tests). */ + forget() { + this.classesAt = 0 + this.empty.clear() + } + + /** pending_classes with waiting releases and not dropped, by their expected rate (a prior for the unexplored). */ + private async classRows(): Promise { + if (Date.now() - this.classesAt < CLASSES_MS) return this.classes + const { rows } = await db.rawQuery( + `SELECT tool, group_id, klass, probed, named, waiting FROM pending_classes WHERE waiting > 0 AND NOT dropped` + ) + this.classes = rows + .map((r: any) => ({ + tool: r.tool as string, + groupId: Number(r.group_id), + klass: Number(r.klass), + probed: Number(r.probed), + named: Number(r.named), + waiting: Number(r.waiting), + })) + .sort((a: ClassRow, b: ClassRow) => classScore(b) - classScore(a)) + this.classesAt = Date.now() + return this.classes + } + + /** + * The next target of up to `n` waiting releases with a target left, claimed: by class, the best record first, + * at most CLASS_SHARE of the claim from one class (the rest explores the classes after it), the cheapest target + * and then the oldest within one; what the classes don't fill, the oldest of the rest (rows from before the + * classes, status 0 of a class without a row). + */ 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, + const out: PendingProbe[] = [] + const perClass = Math.max(1, Math.floor(n * CLASS_SHARE)) + const now = Date.now() + for (const c of await this.classRows()) { + if (out.length >= n) break + const key = `${c.tool}\t${c.groupId}\t${c.klass}` + if ((this.empty.get(key) ?? 0) > now) continue + const want = Math.min(perClass, n - out.length) + const { rows } = await db.rawQuery( + `UPDATE pending_releases p SET status = ? FROM ( + SELECT release_key FROM pending_releases + WHERE status = ? AND tool = ? AND group_id = ? AND klass = ? AND finished AND tries < jsonb_array_length(candidates) + ORDER BY target_bytes NULLS LAST, created_at, first_article 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, p.klass, coalesce(p.poster, '') AS poster, p.target_bytes, p.created_at, p.first_article`, + [PENDING.taken, PENDING.waiting, c.tool, c.groupId, c.klass, want] + ) + if (rows.length < want) this.empty.set(key, now + EMPTY_MS) + if (rows.length > 0) { + await db.rawQuery( + `UPDATE pending_classes SET waiting = greatest(0, waiting - ?) WHERE tool = ? AND group_id = ? AND klass = ?`, + [rows.length, c.tool, c.groupId, c.klass] + ) } - }) + out.push(...claimed(rows).map(toProbe)) + } + if (out.length < n) { + 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, first_article 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, p.klass, coalesce(p.poster, '') AS poster, p.target_bytes, p.created_at, p.first_article`, + [PENDING.taken, PENDING.waiting, n - out.length] + ) + out.push(...claimed(rows).map(toProbe)) + } + return out } /** @@ -106,6 +218,13 @@ class PendingLadder { */ async record(outcomes: PendingOutcome[], trx: TransactionClientContract | typeof db = db) { if (outcomes.length === 0) return + // A twin by content names what the probe itself didn't (hash_twins.ts). + const twins = await twinNames( + outcomes + .filter((o) => o.status === 'done' && o.inspection?.hash16k && o.inspection.size) + .map((o) => ({ hash16k: o.inspection!.hash16k!, size: o.inspection!.size })), + trx + ) const rows = outcomes.map((o) => { const p = o.probe if (o.status === 'waiting') { @@ -118,7 +237,12 @@ class PendingLadder { final: null, } } - const name = o.status === 'done' && o.inspection ? usableName(o.inspection, p.kind) : null + const probed = o.status === 'done' && o.inspection ? usableName(o.inspection, p.kind) : null + const twin = + o.status === 'done' && o.inspection?.hash16k + ? twins.get(twinKey(o.inspection.hash16k, o.inspection.size)) + : undefined + const name = probed ?? twin ?? null const tries = p.tries + 1 const inspection = { kind: p.kind, @@ -202,6 +326,53 @@ class PendingLadder { ), ] ) + // The classes: the same record per (tool, group, class), and a probe given back is waiting again. + const classes = new Map< + string, + { + tool: string + group_id: number + klass: number + probed: number + named: number + waiting: number + } + >() + for (const [i, o] of outcomes.entries()) { + const key = `${o.probe.tool}\t${o.probe.groupId}\t${o.probe.klass}` + const entry = classes.get(key) ?? { + tool: o.probe.tool, + group_id: o.probe.groupId, + klass: o.probe.klass, + probed: 0, + named: 0, + waiting: 0, + } + const final = rows[i].final + if (final) entry.probed++ + if (final === 'named') entry.named++ + if (o.status === 'waiting') entry.waiting++ + classes.set(key, entry) + } + await trx.rawQuery( + `INSERT INTO pending_classes (tool, group_id, klass, probed, named, waiting) + SELECT v.tool, v.group_id, v.klass, v.probed, v.named, v.waiting FROM jsonb_to_recordset(?::jsonb) AS v(tool text, group_id int, klass int, probed int, named int, waiting int) + ON CONFLICT (tool, group_id, klass) DO UPDATE SET probed = pending_classes.probed + excluded.probed, named = pending_classes.named + excluded.named, + waiting = pending_classes.waiting + excluded.waiting, updated_at = now()`, + [JSON.stringify([...classes.values()])] + ) + await trx.rawQuery( + `UPDATE pending_classes c SET dropped = (c.probed >= ?::int AND c.named::float <= c.probed * ?::float) + FROM jsonb_to_recordset(?::jsonb) AS v(tool text, group_id int, klass int) + WHERE c.tool = v.tool AND c.group_id = v.group_id AND c.klass = v.klass`, + [ + pendingMinProbes, + pendingMaxNamed, + JSON.stringify( + [...classes.values()].map(({ tool, group_id, klass }) => ({ tool, group_id, klass })) + ), + ] + ) } /** Waiting releases of a dropped signature discarded without a probe, the explored sample excepted; how many. */ @@ -209,9 +380,10 @@ class PendingLadder { 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 ( + 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, ''))) + OR EXISTS (SELECT 1 FROM pending_classes c WHERE c.dropped AND c.tool = p.tool AND c.group_id = p.group_id AND c.klass = p.klass)) LIMIT ?`, [PENDING.waiting, limit] ) @@ -219,11 +391,32 @@ class PendingLadder { .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)`, + const { rows: gone } = await db.rawQuery( + `UPDATE pending_releases SET status = ? WHERE status = ? AND release_key = ANY(SELECT decode(k, 'hex') FROM unnest(?::text[]) AS k) + RETURNING tool, group_id, klass`, [PENDING.discarded, PENDING.waiting, keys] ) - return rowCount ?? 0 + const counts = new Map() + for (const g of gone) { + const key = `${g.tool}\t${g.group_id}\t${g.klass}` + const entry = counts.get(key) ?? { + tool: g.tool, + group_id: Number(g.group_id), + klass: Number(g.klass), + n: 0, + } + entry.n++ + counts.set(key, entry) + } + if (counts.size > 0) { + await db.rawQuery( + `UPDATE pending_classes c SET waiting = greatest(0, c.waiting - v.n) + FROM jsonb_to_recordset(?::jsonb) AS v(tool text, group_id int, klass int, n int) + WHERE c.tool = v.tool AND c.group_id = v.group_id AND c.klass = v.klass`, + [JSON.stringify([...counts.values()])] + ) + } + return gone.length } } diff --git a/app/services/prober.ts b/app/services/prober.ts index de8a577..d718bc6 100644 --- a/app/services/prober.ts +++ b/app/services/prober.ts @@ -9,6 +9,7 @@ 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 { twinKey, twinNames } from '#services/hash_twins' import PendingLadder, { type PendingProbe } from '#services/pending_ladder' import { PendingNamer } from '#services/pending_namer' @@ -392,10 +393,20 @@ export class Prober { entry.all.push({ kind: r.kind, result: r.result }) byRelease.set(id, entry) } + // A twin by content names a release its own inspections didn't (hash_twins.ts; LOG.md (130)). + const twins = await twinNames( + done.filter((o) => o.inspection!.hash16k && o.inspection!.size).map((o) => ({ hash16k: o.inspection!.hash16k!, size: o.inspection!.size })), + trx + ) + const twinOf = new Map() + for (const o of done) { + const twin = o.inspection!.hash16k ? twins.get(twinKey(o.inspection!.hash16k, o.inspection!.size)) : undefined + if (twin && twin !== byRelease.get(o.candidate.releaseId)?.name) twinOf.set(o.candidate.releaseId, twin) + } const updates = [...byRelease].map(([id, { name, all }]) => ({ id, passworded: all.some((i) => i.result.passworded === true), - inner_name: innerName(all, name), + inner_name: innerName(all, name) ?? twinOf.get(id) ?? null, })) await trx.rawQuery( `UPDATE releases r SET passworded = v.passworded, inner_name = v.inner_name, inspected_at = now() diff --git a/commands/pending_classes.ts b/commands/pending_classes.ts new file mode 100644 index 0000000..85a857b --- /dev/null +++ b/commands/pending_classes.ts @@ -0,0 +1,97 @@ +import { BaseCommand, flags } from '@adonisjs/core/ace' +import type { CommandOptions } from '@adonisjs/core/types/ace' +import db from '@adonisjs/lucid/services/db' +import coordinatorConfig from '#config/coordinator' +import { describeClass } from '#lib/release/index' + +/** + * The pending ladder's classes (LOG.md (130)) on the rows from before them: `klass` (the SQL pending_class) and + * `target_bytes` (the first target's bytes, when its candidates carry them) set in windows of created_at; the + * finished rows among them (named, exhausted, naming) seed each class's record, which gets its verdict (dropped) as + * the ladder gives it; then pending_classes' `waiting` counted afresh from the rows. --report shows the record. + */ +export default class PendingClasses extends BaseCommand { + static commandName = 'pending:classes' + static description = + 'Classes the pending releases from before the classes and recounts pending_classes; --report shows the record' + static options: CommandOptions = { startApp: true } + + @flags.number({ description: 'Hours of created_at per window' }) + declare hours: number + + @flags.boolean({ description: 'Show the classes with the best and worst records, write nothing' }) + declare report: boolean + + async run() { + if (this.report) return this.showReport() + const hours = this.hours ?? 6 + const { rows: span } = await db.rawQuery( + `SELECT min(created_at) AS lo, max(created_at) AS hi FROM pending_releases WHERE klass = 0` + ) + if (!span[0].lo) { + this.logger.info('No rows without a class.') + } else { + let from = new Date(span[0].lo) + const hi = new Date(span[0].hi) + let classed = 0 + const started = Date.now() + while (from <= hi) { + const to = new Date(from.getTime() + hours * 3600_000) + // Rows of the window classed; the finished ones among them (named, exhausted, naming) seed the record. + const { rows } = await db.rawQuery( + `WITH classed AS ( + UPDATE pending_releases SET klass = pending_class(name, bytes, partial), + target_bytes = (SELECT min((c->>'bytes')::bigint) FROM jsonb_array_elements(candidates) c WHERE c->>'bytes' IS NOT NULL) + WHERE klass = 0 AND created_at >= ? AND created_at < ? + RETURNING tool, group_id, klass, status), + seeded AS ( + INSERT INTO pending_classes (tool, group_id, klass, probed, named) + SELECT tool, group_id, klass, count(*), count(*) FILTER (WHERE status IN (1, 4)) FROM classed WHERE status IN (1, 3, 4) GROUP BY 1, 2, 3 + ON CONFLICT (tool, group_id, klass) DO UPDATE SET probed = pending_classes.probed + excluded.probed, named = pending_classes.named + excluded.named + RETURNING 1) + SELECT (SELECT count(*) FROM classed) AS classed, (SELECT count(*) FROM seeded) AS seeded`, + [from.toISOString(), to.toISOString()] + ) + classed += Number(rows[0].classed) + from = to + this.logger.info( + `Up to ${to.toISOString().slice(0, 16)}: ${classed} rows classed (${Math.round((Date.now() - started) / 1000)} s).` + ) + } + // The record's verdicts on the seeded classes. + const { pendingMinProbes, pendingMaxNamed } = coordinatorConfig.probe + await db.rawQuery( + `UPDATE pending_classes SET dropped = true WHERE NOT dropped AND probed >= ?::int AND named::float <= probed * ?::float`, + [pendingMinProbes, pendingMaxNamed] + ) + } + // The waiting counts from the rows themselves (the claim, the discard and the batch store keep them from here). + await db.rawQuery(` + INSERT INTO pending_classes (tool, group_id, klass, waiting) + SELECT tool, group_id, klass, count(*) FROM pending_releases WHERE status = 0 AND finished GROUP BY 1, 2, 3 + ON CONFLICT (tool, group_id, klass) DO UPDATE SET waiting = excluded.waiting`) + await db.rawQuery( + `UPDATE pending_classes c SET waiting = 0 WHERE waiting > 0 AND NOT EXISTS ( + SELECT 1 FROM pending_releases p WHERE p.status = 0 AND p.finished AND p.tool = c.tool AND p.group_id = c.group_id AND p.klass = c.klass)` + ) + const { rows: totals } = await db.rawQuery( + `SELECT count(*) AS classes, coalesce(sum(waiting), 0) AS waiting, count(*) FILTER (WHERE dropped) AS dropped FROM pending_classes` + ) + this.logger.info( + `Done: ${totals[0].classes} classes, ${totals[0].waiting} waiting, ${totals[0].dropped} dropped.` + ) + } + + private async showReport() { + const { rows } = await db.rawQuery( + `SELECT c.tool, g.name AS grp, c.klass, c.probed, c.named, c.waiting, c.dropped FROM pending_classes c JOIN groups g ON g.id = c.group_id + WHERE c.probed >= 20 ORDER BY c.named::float / c.probed DESC, c.probed DESC` + ) + const line = (r: any) => + ` ${r.tool.padEnd(10)} ${r.grp.padEnd(32)} ${describeClass(Number(r.klass)).padEnd(26)} probed ${String(r.probed).padStart(7)} named ${String(Math.round((100 * r.named) / r.probed)).padStart(3)}% waiting ${String(r.waiting).padStart(8)}${r.dropped ? ' dropped' : ''}` + this.logger.info(`${rows.length} classes with 20+ climbs; the best:`) + for (const r of rows.slice(0, 25)) this.logger.info(line(r)) + this.logger.info('The worst:') + for (const r of rows.slice(-25)) this.logger.info(line(r)) + } +} diff --git a/commands/releases_readable.ts b/commands/releases_readable.ts index bd23674..556879d 100644 --- a/commands/releases_readable.ts +++ b/commands/releases_readable.ts @@ -9,7 +9,8 @@ const CURSOR = 'readable' /** * The readable flag of the catalog's releases by today's classifier (isUnreadableName, LOG.md (129)): rows flagged * readable whose name no one can read ("Zx5AjNaK-mh7CFg6Hf8uu", "Description - Lt5nHjlo3RiuBb3C") get readable = - * false, so the readable filter hides them (an inner name from the prober still shows them). By the release name, + * false, so the readable filter hides them (an inner name from the prober still shows them), and their probe + * candidates marked unreadable, the prober's first tiers. By the release name, * which is the shared file stem for most rows; not by the chunks' file names, which the edge judges from now on. * Batches by id, a dry run by default; --apply writes, and goes on from its cursor when run again. */ @@ -59,10 +60,16 @@ export default class ReleasesReadable extends BaseCommand { } changed += ids.length if (this.apply) { - if (ids.length > 0) + if (ids.length > 0) { await db.rawQuery(`UPDATE releases SET readable = false WHERE id = ANY(?::bigint[])`, [ ids, ]) + // Their probe candidates to the unreadable tiers, which the prober takes first (LOG.md (130)). + await db.rawQuery( + `UPDATE probe_candidates SET unreadable = true WHERE release_id = ANY(?::bigint[]) AND NOT unreadable`, + [ids] + ) + } await db.rawQuery( `INSERT INTO worker_cursors (name, at, key) VALUES (?, now(), ?) ON CONFLICT (name) DO UPDATE SET at = excluded.at, key = excluded.key`, [CURSOR, String(to)] diff --git a/commands/rules_classes.ts b/commands/rules_classes.ts new file mode 100644 index 0000000..b356a6a --- /dev/null +++ b/commands/rules_classes.ts @@ -0,0 +1,54 @@ +import { BaseCommand, flags } from '@adonisjs/core/ace' +import type { CommandOptions } from '@adonisjs/core/types/ace' +import db from '@adonisjs/lucid/services/db' +import RuleSets from '#services/rule_sets' +import { classKey, describeClass } from '#lib/release/index' + +/** + * Publishes a rule set whose dropped classes are the pending ladder's (pending_classes.dropped; LOG.md (130)): the + * edge keeps unnamed releases of those classes out of the pending store from then on, their explored sample + * excepted. Run again as the record grows; unchanged classes publish nothing. + */ +export default class RulesClasses extends BaseCommand { + static commandName = 'rules:classes' + static description = "Publishes a rule set with the pending ladder's dropped classes" + static options: CommandOptions = { startApp: true } + + @flags.boolean({ description: 'Show what would be published, write nothing' }) + declare dryRun: boolean + + async run() { + const { rows } = await db.rawQuery( + `SELECT c.tool, g.name AS grp, c.klass, c.probed, c.named FROM pending_classes c JOIN groups g ON g.id = c.group_id WHERE c.dropped ORDER BY 1, 2, 3` + ) + const next = new Set(rows.map((r: any) => classKey(r.tool, r.grp, Number(r.klass)))) + const { version, priorities: _p, ...current } = await RuleSets.current() + const before = new Set(current.droppedClasses ?? []) + const added = [...next].filter((c) => !before.has(c)) + const removed = [...before].filter((c) => !next.has(c)) + if (added.length === 0 && removed.length === 0) { + this.logger.info( + `Unchanged: rule set ${version} has the ${before.size} dropped classes already.` + ) + return + } + for (const r of rows.slice(0, 40)) { + this.logger.info( + ` ${r.tool.padEnd(10)} ${r.grp.padEnd(32)} ${describeClass(Number(r.klass)).padEnd(26)} ${r.named} of ${r.probed} named` + ) + } + if (this.dryRun) { + this.logger.info( + `Would publish ${next.size} dropped classes (${added.length} added, ${removed.length} removed).` + ) + return + } + const published = await RuleSets.publish( + { ...current, droppedClasses: [...next].sort() }, + `${next.size} dropped classes from the pending ladder` + ) + this.logger.info( + `Rule set ${published}: ${next.size} dropped classes (${added.length} added, ${removed.length} removed).` + ) + } +} diff --git a/database/migrations/1770000006800_pending_classes.ts b/database/migrations/1770000006800_pending_classes.ts new file mode 100644 index 0000000..f0204d4 --- /dev/null +++ b/database/migrations/1770000006800_pending_classes.ts @@ -0,0 +1,57 @@ +import { BaseSchema } from '@adonisjs/lucid/schema' + +/** + * The pending ladder by class (REBUILD.md decision f; LOG.md (130)): a pending release's class (lib/release/classes.ts: + * size bucket × name shape × partial, 1..70; 0 for rows from before) and the bytes of its first target, so the + * claim takes the classes with the best record first and the cheapest article within one; pending_classes per + * (tool, group, class) how many climbed to the end and how many got a name, `waiting` how many wait (kept by the + * batch store, the claim and the discard; recounted by pending:classes), `dropped` as pending_signatures. + * The SQL pending_class() is the TypeScript's pendingClass, for the batch store's insert and the backfill. + */ +export default class extends BaseSchema { + static disableTransactions = true + + async up() { + this.schema.raw(` + CREATE OR REPLACE FUNCTION pending_class(name text, bytes bigint, partial boolean) RETURNS smallint LANGUAGE sql IMMUTABLE AS $$ + SELECT ((CASE WHEN bytes < 1000000 THEN 0 WHEN bytes < 10000000 THEN 1 WHEN bytes < 100000000 THEN 2 WHEN bytes < 1000000000 THEN 3 ELSE 4 END) * 14 + + (CASE WHEN s ~ '^[0-9a-fA-F]{16,}$' THEN 1 + WHEN s ~ '^[0-9]{6,}$' THEN 2 + WHEN s ~ '^[a-z2-7]{20,}$' AND s ~ '[2-7]' THEN 3 + WHEN s ~ '^[A-Za-z]+$' THEN 4 + WHEN s ~ '^[A-Za-z0-9]+$' THEN 5 + WHEN s ~ '^[A-Za-z0-9._-]+$' THEN 6 + ELSE 7 END - 1) * 2 + + (CASE WHEN partial THEN 1 ELSE 0 END) + 1)::smallint + FROM (SELECT regexp_replace(name, '\\.[A-Za-z0-9]{1,4}$', '') AS s) x + $$`) + this.schema.raw( + `ALTER TABLE pending_releases ADD COLUMN IF NOT EXISTS klass smallint NOT NULL DEFAULT 0` + ) + this.schema.raw(`ALTER TABLE pending_releases ADD COLUMN IF NOT EXISTS target_bytes bigint`) + this.schema.raw(` + CREATE TABLE IF NOT EXISTS pending_classes ( + tool varchar(16) NOT NULL, + group_id integer NOT NULL, + klass smallint NOT NULL, + probed integer NOT NULL DEFAULT 0, + named integer NOT NULL DEFAULT 0, + waiting integer NOT NULL DEFAULT 0, + dropped boolean NOT NULL DEFAULT false, + updated_at timestamptz NOT NULL DEFAULT now(), + PRIMARY KEY (tool, group_id, klass) + )`) + // The claim by class: the cheapest target first within one, the oldest among equals. + this.schema.raw( + `CREATE INDEX CONCURRENTLY IF NOT EXISTS pending_releases_class_idx ON pending_releases (tool, group_id, klass, target_bytes, created_at) WHERE status = 0` + ) + } + + async down() { + this.schema.raw(`DROP INDEX CONCURRENTLY IF EXISTS pending_releases_class_idx`) + this.schema.raw(`DROP TABLE IF EXISTS pending_classes`) + this.schema.raw(`ALTER TABLE pending_releases DROP COLUMN IF EXISTS target_bytes`) + this.schema.raw(`ALTER TABLE pending_releases DROP COLUMN IF EXISTS klass`) + this.schema.raw(`DROP FUNCTION IF EXISTS pending_class(text, bigint, boolean)`) + } +} diff --git a/edge/src/indexer.rs b/edge/src/indexer.rs index 1ff2cba..2d16a28 100644 --- a/edge/src/indexer.rs +++ b/edge/src/indexer.rs @@ -72,6 +72,8 @@ pub struct Candidate { pub kind: &'static str, pub message_id: String, pub file: String, + /// The target article's bytes: the cheapest probe first within a class (LOG.md (130)). + pub bytes: i64, } #[derive(Default)] @@ -1451,7 +1453,16 @@ impl RangeIndexer { } if !chunk.readable { // Unnamed: no chunk (its segments are read again once a probe names it), the tool for its signature. - chunk.tool = Some(names::upload_tool(&list[0][0].message_id)); + let tool = names::upload_tool(&list[0][0].message_id); + // A class the ladder found never names anything stays out of the pending store, its explored + // sample excepted (so the rate keeps being measured); counted as garbage (LOG.md (130)). + let klass = names::pending_class(&chunk.name, chunk.bytes.max(0) as u64, chunk.partial); + if self.rules.dropped_classes.contains(&names::class_key(tool, &self.group, klass)) && !names::explored(key, self.rules.sample_percent) { + let segments: usize = list.iter().map(|f| f.len()).sum(); + self.count("garbage", format!("{tool} {} #{klass}", self.group), segments as u64, list.len() as u64); + return; + } + chunk.tool = Some(tool); chunk.payload = encode_chunk(&[]); } if chunk.readable { @@ -1574,7 +1585,7 @@ fn chunk_of(key: &str, files: &[Vec], upload: usize) -> ReleaseChunk { fn candidates_of(key: &str, files: &[Vec], upload: usize) -> Vec { let mut out = Vec::new(); let mut tail: Option<(i64, &Vec)> = None; - let c = |kind, s: &Seg| Candidate { release_key: key.to_string(), upload, kind, message_id: s.message_id.clone(), file: s.file.name.clone() }; + let c = |kind, s: &Seg| Candidate { release_key: key.to_string(), upload, kind, message_id: s.message_id.clone(), file: s.file.name.clone(), bytes: s.bytes }; for f in files { let name = &f[0].file.name; let volume = names::seven_zip_volume(name); @@ -1595,7 +1606,7 @@ fn candidates_of(key: &str, files: &[Vec], upload: usize) -> Vec } } if let Some((_, f)) = tail { - out.push(Candidate { release_key: key.to_string(), upload, kind: "7z-tail", message_id: f.last().unwrap().message_id.clone(), file: f[0].file.name.clone() }); + out.push(Candidate { release_key: key.to_string(), upload, kind: "7z-tail", message_id: f.last().unwrap().message_id.clone(), file: f[0].file.name.clone(), bytes: f.last().unwrap().bytes }); } if out.is_empty() && files.len() == 1 && is_random_file(&files[0][0].file.name) { out.push(c("lone", &files[0][0])); diff --git a/edge/src/lease.rs b/edge/src/lease.rs index f5c2ef7..44e0adb 100644 --- a/edge/src/lease.rs +++ b/edge/src/lease.rs @@ -62,6 +62,8 @@ pub struct RulesJson { #[serde(default)] pub dropped_signatures: Vec, #[serde(default)] + pub dropped_classes: Vec, + #[serde(default)] pub dropped_tools: Option>, #[serde(default)] pub per_article_max_parts: Option, @@ -83,6 +85,7 @@ impl RulesJson { let mut rules = Rules::defaults(self.priorities.into_iter().collect()); rules.version = self.version; rules.dropped_signatures = self.dropped_signatures.into_iter().collect(); + rules.dropped_classes = self.dropped_classes.into_iter().collect(); if let Some(tools) = self.dropped_tools { rules.dropped_tools = tools.into_iter().collect(); } diff --git a/edge/src/names.rs b/edge/src/names.rs index 9ffa9be..c530554 100644 --- a/edge/src/names.rs +++ b/edge/src/names.rs @@ -1228,6 +1228,39 @@ pub fn is_unreadable_name(name: &str) -> bool { random > 0 && random > words } +/// The class of an unnamed release (lib/release/classes.ts pendingClass; LOG.md (130)): size bucket × name shape × +/// partial, 1..70. The lease's dropped classes ("tool\tgroup\tklass") are skipped at the edge. +pub fn pending_class(name: &str, bytes: u64, partial: bool) -> u8 { + let size = [1_000_000u64, 10_000_000, 100_000_000, 1_000_000_000].iter().filter(|&&b| bytes >= b).count() as u8; + let s = match name.rfind('.') { + Some(i) if (1..=4).contains(&(name.len() - i - 1)) && name[i + 1..].bytes().all(|b| b.is_ascii_alphanumeric()) => &name[..i], + _ => name, + }; + let b = s.as_bytes(); + let all = |f: fn(&u8) -> bool| !b.is_empty() && b.iter().all(f); + let shape: u8 = if b.len() >= 16 && all(u8::is_ascii_hexdigit) { + 1 + } else if b.len() >= 6 && all(u8::is_ascii_digit) { + 2 + } else if b.len() >= 20 && all(|c| c.is_ascii_lowercase() || (b'2'..=b'7').contains(c)) && b.iter().any(|c| (b'2'..=b'7').contains(c)) { + 3 + } else if all(u8::is_ascii_alphabetic) { + 4 + } else if all(u8::is_ascii_alphanumeric) { + 5 + } else if all(|c| c.is_ascii_alphanumeric() || matches!(c, b'.' | b'_' | b'-')) { + 6 + } else { + 7 + }; + size * 14 + (shape - 1) * 2 + u8::from(partial) + 1 +} + +/// A class as droppedClasses lists it. +pub fn class_key(tool: &str, group: &str, klass: u8) -> String { + format!("{tool}\t{group}\t{klass}") +} + /// looksScrambled: at least half the letters and digits of a file's stem in tokens of 5 or more that switch case /// (lower to upper) or between letters and digits 3 times or more ("nYDeXeF49dieZ1J7flK-JFzZ3j.ZEDsr", "bslXlwgPiqZadhRa"). /// The names a random-subject-per-article upload gives each article that looksRandom misses for their separators. @@ -1559,6 +1592,8 @@ pub struct Rules { pub priorities: rustc_hash::FxHashMap, /// "tool\tgroup" pub dropped_signatures: rustc_hash::FxHashSet, + /// "tool\tgroup\tklass" (pending_class): unnamed releases of these classes don't go to the pending store. + pub dropped_classes: rustc_hash::FxHashSet, pub dropped_tools: rustc_hash::FxHashSet, pub per_article_max_parts: u32, pub sample_percent: f64, @@ -1577,6 +1612,7 @@ impl Rules { version: 1, priorities, dropped_signatures: Default::default(), + dropped_classes: Default::default(), dropped_tools: ["ngPost", "hex32", "yuun", "JBinUp", "random"].iter().map(|t| t.to_string()).collect(), per_article_max_parts: 3, sample_percent: 1.0, @@ -1785,6 +1821,20 @@ mod tests { assert_eq!(id(r#"[2/6] - "x.jpg" yEnc"#, "x.jpg").key, id(r#"[1/6] - "x.part01.rar" yEnc"#, "x.part01.rar").key); } + #[test] + fn pending_classes() { + use super::pending_class; + // Size bucket × shape × partial: a 5 MB hex name whole, the same partial, a 500 MB scramble with separators. + assert_eq!(pending_class("c04212a389cf69edc54767355ab5285f", 5_000_000, false), 15); + assert_eq!(pending_class("c04212a389cf69edc54767355ab5285f", 5_000_000, true), 16); + assert_eq!(pending_class("Zx5AjNaK-mh7CFg6Hf8uu", 500_000_000, false), 3 * 14 + 5 * 2 + 1); + assert_eq!(pending_class("1650777167", 100, false), 3); + assert_eq!(pending_class("mrrtmzrzmi4dgmrzmyytqmlbztczbx", 2_000_000_000, false), 4 * 14 + 2 * 2 + 1); + assert_eq!(pending_class("PsNdnYiugfZ.rar", 20_000_000, false), 2 * 14 + 3 * 2 + 1); + assert_eq!(pending_class("n5d1ueIf34X", 20_000_000, true), 2 * 14 + 4 * 2 + 2); + assert_eq!(pending_class("sgkra vU7u2yuw9LDLCjDu04pOez", 999, false), 13); + } + #[test] fn unreadable_names() { use super::is_unreadable_name; diff --git a/edge/src/wire.rs b/edge/src/wire.rs index b2a7e9c..33d6d17 100644 --- a/edge/src/wire.rs +++ b/edge/src/wire.rs @@ -44,6 +44,7 @@ struct WireCandidate<'a> { kind: &'a str, message_id: &'a str, file: &'a str, + bytes: i64, } #[derive(Serialize)] @@ -97,7 +98,7 @@ fn release<'a>(r: &'a ReleaseChunk, rule: Option<(&'a str, &'a str)>) -> WireRel } fn candidate(c: &Candidate) -> WireCandidate<'_> { - WireCandidate { release_key: &c.release_key, upload: c.upload, kind: c.kind, message_id: &c.message_id, file: &c.file } + WireCandidate { release_key: &c.release_key, upload: c.upload, kind: c.kind, message_id: &c.message_id, file: &c.file, bytes: c.bytes } } /// toWire(batch) as JSON bytes. diff --git a/lib/edge/range_indexer.ts b/lib/edge/range_indexer.ts index 59a2cfc..40e6620 100644 --- a/lib/edge/range_indexer.ts +++ b/lib/edge/range_indexer.ts @@ -2,7 +2,8 @@ import { hash } from 'node:crypto' import type { Overview } from '../nntp/index.js' import { FileAggregator, type AggregatedFile, isArchiveStart, isLaterVolume, sevenZipVolume } from '../files/index.js' import { RuleSet, isRandomFile, isScrambledFile, uploadTool, type Rules } from '../rules/index.js' -import { isUnreadableName } from '../release/index.js' +import { classKey, isUnreadableName, pendingClass } from '../release/index.js' +import { explored } from '../rules/upload_tools.js' import { SegmentScanner, scanOverviewText, type EdgeSegment } from './scan.js' import { encodeChunk, type ChunkFile } from '../chunks/index.js' @@ -59,6 +60,8 @@ export interface ProbeCandidate { kind: CandidateKind messageId: string file: string + /** The target article's bytes: the cheapest probe first within a class (LOG.md (130)). */ + bytes?: number } /** What a rule dropped in a batch (REBUILD.md, rules and recovery). */ @@ -263,7 +266,18 @@ export class RangeIndexer { true ) if (verdict.rule === null) { - this.batch.releases.push(chunkOf(key, list, upload)) + const chunk = chunkOf(key, list, upload) + if (!chunk.readable) { + // Unnamed: a class the ladder found never names anything stays out of the pending store, its explored + // sample excepted (so the rate keeps being measured); counted as garbage (LOG.md (130)). + const tool = uploadTool(list[0].segments[0].messageId) + const klass = pendingClass(chunk.name, chunk.bytes, chunk.partial === true) + if (this.rules.classes.has(classKey(tool, this.options.group, klass)) && !explored(key, this.options.rules.samplePercent)) { + this.count('garbage', `${tool} ${this.options.group} #${klass}`, list.reduce((n, f) => n + f.segments.length, 0), list.length) + return + } + } + this.batch.releases.push(chunk) this.batch.candidates.push(...candidatesOf(key, list, upload)) return } @@ -483,12 +497,12 @@ export function candidatesOf(key: string, files: AggregatedFile[], } // An unsplit 7z of more than one segment: its listing is in its last one. if (volume === 0 && f.segments.length > 1 && !tail) tail = { volume, file: f } - if (isArchiveStart(name)) out.push({ releaseKey: key, upload, kind: 'archive', messageId: f.segments[0].messageId, file: name }) - else if (/\.par2$/i.test(name) && !isLaterVolume(name)) out.push({ releaseKey: key, upload, kind: 'par2', messageId: f.segments[0].messageId, file: name }) + if (isArchiveStart(name)) out.push({ releaseKey: key, upload, kind: 'archive', messageId: f.segments[0].messageId, file: name, bytes: f.segments[0].bytes }) + else if (/\.par2$/i.test(name) && !isLaterVolume(name)) out.push({ releaseKey: key, upload, kind: 'par2', messageId: f.segments[0].messageId, file: name, bytes: f.segments[0].bytes }) } - if (tail) out.push({ releaseKey: key, upload, kind: '7z-tail', messageId: tail.file.segments.at(-1)!.messageId, file: tail.file.segments[0].file.name }) + if (tail) out.push({ releaseKey: key, upload, kind: '7z-tail', messageId: tail.file.segments.at(-1)!.messageId, file: tail.file.segments[0].file.name, bytes: tail.file.segments.at(-1)!.bytes }) if (out.length === 0 && files.length === 1 && isRandomFile(files[0].segments[0].file.name)) { - out.push({ releaseKey: key, upload, kind: 'lone', messageId: files[0].segments[0].messageId, file: files[0].segments[0].file.name }) + out.push({ releaseKey: key, upload, kind: 'lone', messageId: files[0].segments[0].messageId, file: files[0].segments[0].file.name, bytes: files[0].segments[0].bytes }) } return out.slice(0, MAX_CANDIDATES) } diff --git a/lib/release/classes.ts b/lib/release/classes.ts new file mode 100644 index 0000000..0b7c7dc --- /dev/null +++ b/lib/release/classes.ts @@ -0,0 +1,55 @@ +/** + * The class of an unnamed release (REBUILD.md decision f; LOG.md (130)): what the headers say about its chance of a + * name, from the pending ladder's own record. Size bucket × name shape × partial, 1..70, kept in step with + * names.rs pending_class and the SQL pending_class() (migration 6800): the batch store sets it with the SQL, the + * edge skips the classes a lease marks dropped. + */ + +/** Size buckets: under 1 MB, 10 MB, 100 MB, 1 GB, and from 1 GB. */ +export const SIZE_BUCKETS = [1_000_000, 10_000_000, 100_000_000, 1_000_000_000] +/** Name shapes, 1..7: hex, digits, base32, letters only, mixed letters and digits, with separators, other. */ +export const SHAPES = [ + 'hex', + 'digits', + 'base32', + 'letters', + 'alnum', + 'separators', + 'other', +] as const + +export function sizeBucket(bytes: number): number { + let i = 0 + while (i < SIZE_BUCKETS.length && bytes >= SIZE_BUCKETS[i]) i++ + return i +} + +export function nameShape(name: string): number { + const s = name.replace(/\.[A-Za-z0-9]{1,4}$/, '') + if (/^[0-9a-fA-F]{16,}$/.test(s)) return 1 + if (/^[0-9]{6,}$/.test(s)) return 2 + if (/^[a-z2-7]{20,}$/.test(s) && /[2-7]/.test(s)) return 3 + if (/^[A-Za-z]+$/.test(s)) return 4 + if (/^[A-Za-z0-9]+$/.test(s)) return 5 + if (/^[A-Za-z0-9._-]+$/.test(s)) return 6 + return 7 +} + +/** 1..70: (size bucket × 7 + shape − 1) × 2 + partial + 1. */ +export function pendingClass(name: string, bytes: number, partial: boolean): number { + return sizeBucket(bytes) * 14 + (nameShape(name) - 1) * 2 + (partial ? 1 : 0) + 1 +} + +/** A class as droppedClasses lists it: "tool\tgroup\tklass". */ +export function classKey(tool: string, group: string, klass: number): string { + return `${tool}\t${group}\t${klass}` +} + +/** A class in words ("10-100M hex partial"). */ +export function describeClass(klass: number): string { + const k = klass - 1 + const size = Math.floor(k / 14) + const shape = SHAPES[Math.floor((k % 14) / 2)] + const sizes = ['<1M', '1-10M', '10-100M', '100M-1G', '1G+'] + return `${sizes[size]} ${shape}${k % 2 ? ' partial' : ''}` +} diff --git a/lib/release/index.ts b/lib/release/index.ts index 23fdb54..881ce1e 100644 --- a/lib/release/index.ts +++ b/lib/release/index.ts @@ -4,3 +4,4 @@ export { releaseIdentity, displayName, par2ReleaseIdentity, par2ReleaseName, rarReleaseIdentity, isGenericName, fileStem, sizeAfterName, runOn, subjectRemainder, namesRelease, isPostToken, type ReleaseIdentity, type ReleaseInput } from './identity.js' export { looksRandom } from './random.js' export { isUnreadableName, tokenKind } from './readable.js' +export { pendingClass, classKey, describeClass, nameShape, sizeBucket } from './classes.js' diff --git a/lib/rules/rule_set.ts b/lib/rules/rule_set.ts index 2169eda..f1e36b4 100644 --- a/lib/rules/rule_set.ts +++ b/lib/rules/rule_set.ts @@ -12,6 +12,11 @@ export interface Rules { priorities: Record /** Dropped signatures, as "tool\tgroup" (GarbageFilter's table). */ droppedSignatures: string[] + /** + * Dropped classes of unnamed releases, as "tool\tgroup\tklass" (lib/release/classes.ts; the pending ladder's record, + * LOG.md (130)): the edge keeps them out of the pending store, the explored sample excepted. + */ + droppedClasses: string[] /** Tools whose random-named uploads are dropped in every group (named < 1%; LOG.md 2026-10-08 (5)). */ droppedTools: string[] /** A random-named file of at most this many parts: a random subject per article, needs bodies. */ @@ -53,6 +58,7 @@ export const DEFAULT_RULES: Omit = { droppedPosters: ['celebmovie', 'celebwow', 'celebfx', 'iluvcelebs', 'nudecelebs', 'celebsextapes'], dropStubArchives: true, dropNonsensePieces: true, + droppedClasses: [], } /** The stub-archive rule's bounds (edge/src/names.rs): Power-Post's 250 kB default article, the fakes' size span. */ @@ -178,11 +184,13 @@ export function ownerOf(groups: string[], priorities: Record): s export class RuleSet { private signatures: Set + readonly classes: Set private tools: Set private posters: string[] constructor(readonly rules: Rules) { this.posters = rules.droppedPosters ?? [] this.signatures = new Set(rules.droppedSignatures) + this.classes = new Set(rules.droppedClasses ?? []) this.tools = new Set(rules.droppedTools) } diff --git a/tests/functional/pending_classes.spec.ts b/tests/functional/pending_classes.spec.ts new file mode 100644 index 0000000..6314351 --- /dev/null +++ b/tests/functional/pending_classes.spec.ts @@ -0,0 +1,28 @@ +import { test } from '@japa/runner' +import db from '@adonisjs/lucid/services/db' +import { pendingClass } from '#lib/release/index' + +test.group('Pending classes / SQL pending_class()', () => { + test('the SQL function agrees with pendingClass', async ({ assert }) => { + const cases: [string, number, boolean][] = [ + ['c04212a389cf69edc54767355ab5285f', 5_000_000, false], + ['c04212a389cf69edc54767355ab5285f', 5_000_000, true], + ['Zx5AjNaK-mh7CFg6Hf8uu', 500_000_000, false], + ['1650777167', 100, false], + ['mrrtmzrzmi4dgmrzmyytqmlbztczbx', 2_000_000_000, false], + ['PsNdnYiugfZ.rar', 20_000_000, false], + ['n5d1ueIf34X', 20_000_000, true], + ['sgkra vU7u2yuw9LDLCjDu04pOez', 999, false], + ['Description - Lt5nHjlo3RiuBb3C', 40_000_000, false], + ['ABCDEFGHIJKLMNOP', 0, false], + ] + for (const [name, bytes, partial] of cases) { + const { rows } = await db.rawQuery(`SELECT pending_class(?, ?::bigint, ?) AS k`, [ + name, + bytes, + partial, + ]) + assert.equal(Number(rows[0].k), pendingClass(name, bytes, partial), name) + } + }) +}) diff --git a/tests/functional/pending_ladder.spec.ts b/tests/functional/pending_ladder.spec.ts index a275ed1..e85a1cd 100644 --- a/tests/functional/pending_ladder.spec.ts +++ b/tests/functional/pending_ladder.spec.ts @@ -9,7 +9,7 @@ let n = 0 async function pending( name: string, - candidates: { kind: string; messageId: string; file: string }[], + candidates: { kind: string; messageId: string; file: string; bytes?: number }[], extra: Record = {} ) { n++ @@ -206,3 +206,198 @@ test.group('Pending ladder', (group) => { assert.lengthOf(await db.from('pending_signatures'), 0) }) }) + +test.group('Pending ladder / classes', (group) => { + group.each.setup(async () => { + PendingLadder.forget() + const rollback = await testUtils.db().withGlobalTransaction() + groupId = Number( + ( + await db + .table('groups') + .insert({ name: 'alt.binaries.classes', priority: 1 }) + .returning('id') + )[0].id + ) + n = 0 + return rollback + }) + + test('a claim takes the class with the best record first, the cheapest target within it, the rest oldest first', async ({ + assert, + }) => { + // Class 15 (1-10 MB hex, whole) named 9 of 10; class 16 (partial) named 0 of 50; class 0: rows from before. + await db.table('pending_classes').multiInsert([ + { tool: 'other', group_id: groupId, klass: 15, probed: 10, named: 9, waiting: 2 }, + { tool: 'other', group_id: groupId, klass: 16, probed: 50, named: 0, waiting: 1 }, + ]) + const old = await pending('00000000000000000000000000000031', [ + { kind: 'lone', messageId: '', file: 'o' }, + ]) + const poor = await pending( + '00000000000000000000000000000032', + [{ kind: 'lone', messageId: '', file: 'p' }], + { klass: 16 } + ) + const bigTarget = await pending( + '00000000000000000000000000000033', + [{ kind: 'archive', messageId: '', file: 'b.rar', bytes: 700_000 }], + { klass: 15, target_bytes: 700_000 } + ) + const smallTarget = await pending( + '00000000000000000000000000000034', + [{ kind: 'archive', messageId: '', file: 's.rar', bytes: 50_000 }], + { klass: 15, target_bytes: 50_000 } + ) + const probes = await PendingLadder.claim(10) + assert.deepEqual( + probes.map((p) => [p.releaseKey, p.klass]), + [ + [smallTarget, 15], + [bigTarget, 15], + [poor, 16], + [old, 0], + ] + ) + const waiting = await db + .from('pending_classes') + .where({ group_id: groupId }) + .orderBy('klass') + .select('klass', 'waiting') + assert.deepEqual( + waiting.map((c) => [c.klass, c.waiting]), + [ + [15, 0], + [16, 0], + ] + ) + // Outcomes count on the class; a probe given back waits again. + await PendingLadder.record([ + { + probe: probes[0], + status: 'done', + inspection: inspection('s.rar', ['Some.Movie.2016.1080p.mkv']), + }, + { probe: probes[1], status: 'waiting' }, + { probe: probes[2], status: 'missing' }, + ]) + const classes = await db.from('pending_classes').where({ group_id: groupId }).orderBy('klass') + assert.deepEqual( + classes.map((c) => [c.klass, c.probed, c.named, c.waiting, c.dropped]), + [ + [15, 11, 10, 1, false], + [16, 51, 0, 0, false], + ] + ) + }) + + test('a class that names nothing is dropped and its waiting releases discarded, the other classes untouched', async ({ + assert, + }) => { + await db + .table('pending_classes') + .insert({ tool: 'other', group_id: groupId, klass: 16, probed: 99, named: 0, waiting: 1 }) + const last = await pending( + '00000000000000000000000000000041', + [{ kind: 'lone', messageId: '', file: 'z' }], + { klass: 16 } + ) + const [probe] = await PendingLadder.claim(1) + assert.equal(probe.releaseKey, last) + await PendingLadder.record([{ probe, status: 'missing' }]) + assert.include( + await db.from('pending_classes').where({ group_id: groupId, klass: 16 }).firstOrFail(), + { probed: 100, named: 0, dropped: true } + ) + const waiting = await pending( + '00000000000000000000000000000042', + [{ kind: 'lone', messageId: '', file: 'w' }], + { klass: 16 } + ) + const other = await pending( + '00000000000000000000000000000043', + [{ kind: 'lone', messageId: '', file: 'o' }], + { klass: 15 } + ) + await db.rawQuery( + `UPDATE pending_classes SET waiting = waiting + 1 WHERE group_id = ? AND klass = 16`, + [groupId] + ) + await db.table('rule_sets').insert({ + version: 98, + 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) + assert.equal( + (await db.from('pending_classes').where({ group_id: groupId, klass: 16 }).firstOrFail()) + .waiting, + 0 + ) + }) + + test('a twin by content names a release whose own probe did not', async ({ assert }) => { + // A catalog release with a readable name holds the same first 16 KiB and size. + const [poster] = await db.table('posters').insert({ poster: 'Twin ' }).returning('id') + const [set] = await db + .table('group_sets') + .insert({ groups: JSON.stringify(['alt.binaries.classes']) }) + .returning('id') + const [twin] = await db + .table('releases') + .insert({ + release_key: Buffer.from('aa'.repeat(20), 'hex'), + identity_key: Buffer.from('bb'.repeat(20), 'hex'), + group_id: groupId, + group_set_id: set.id, + poster_id: poster.id, + name: 'Some.Movie.2016.1080p.BluRay-GRP', + source: 3, + posted_at: new Date('2026-10-01T00:00:00Z'), + first_article: 1, + last_article: 2, + files: 2, + segments: 2, + bytes: 1000, + readable: true, + }) + .returning('id') + await db.table('file_hashes').insert({ + candidate_id: 777_001, + release_id: twin.id, + hash16k: 'deadbeef'.repeat(4), + size: 1000, + }) + const key = await pending('00000000000000000000000000000051', [ + { kind: 'lone', messageId: '', file: '00000000000000000000000000000051' }, + ]) + const [probe] = await PendingLadder.claim(1) + assert.equal(probe.releaseKey, key) + await PendingLadder.record([ + { + probe, + status: 'done', + inspection: { + ...inspection('00000000000000000000000000000051'), + hash16k: 'deadbeef'.repeat(4), + }, + }, + ]) + assert.include( + await db.from('pending_releases').where('release_key', Buffer.from(key, 'hex')).firstOrFail(), + { + status: PENDING.naming, + named: 'Some.Movie.2016.1080p.BluRay-GRP', + } + ) + }) +}) diff --git a/tests/unit/release/classes.spec.ts b/tests/unit/release/classes.spec.ts new file mode 100644 index 0000000..7f5bc09 --- /dev/null +++ b/tests/unit/release/classes.spec.ts @@ -0,0 +1,22 @@ +import { test } from '@japa/runner' +import { classKey, describeClass, pendingClass } from '#lib/release/index' + +test.group('release / pendingClass', () => { + test('size bucket × name shape × partial, as names.rs pending_class', ({ assert }) => { + // The same cases as the Rust test. + assert.equal(pendingClass('c04212a389cf69edc54767355ab5285f', 5_000_000, false), 15) + assert.equal(pendingClass('c04212a389cf69edc54767355ab5285f', 5_000_000, true), 16) + assert.equal(pendingClass('Zx5AjNaK-mh7CFg6Hf8uu', 500_000_000, false), 3 * 14 + 5 * 2 + 1) + assert.equal(pendingClass('1650777167', 100, false), 3) + assert.equal( + pendingClass('mrrtmzrzmi4dgmrzmyytqmlbztczbx', 2_000_000_000, false), + 4 * 14 + 2 * 2 + 1 + ) + assert.equal(pendingClass('PsNdnYiugfZ.rar', 20_000_000, false), 2 * 14 + 3 * 2 + 1) + assert.equal(pendingClass('n5d1ueIf34X', 20_000_000, true), 2 * 14 + 4 * 2 + 2) + assert.equal(pendingClass('sgkra vU7u2yuw9LDLCjDu04pOez', 999, false), 13) + assert.equal(describeClass(16), '1-10M hex partial') + assert.equal(describeClass(3 * 14 + 5 * 2 + 1), '100M-1G separators') + assert.equal(classKey('nyuu', 'alt.binaries.boneless', 16), 'nyuu\talt.binaries.boneless\t16') + }) +})