From 3d0dc6ca1f86e93eb383986b44282dad23ec2775 Mon Sep 17 00:00:00 2001 From: Daan Meijer Date: Mon, 12 Oct 2026 09:37:35 +0200 Subject: [PATCH] The par2 merge: releases whose files one par2 set describes become one MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The prober stores, per probed article that starts a file, the MD5 of its first 16 KiB (file_hashes), and per par2 body the set's whole file list (par2_files, par2_sets): FileDesc holds the same hash per file, whatever the file was renamed to. releases:merge (the merger service, no NNTP) merges every minute the releases of one group, posted within 48 hours of a release holding the set's par2, whose files the set describes, into the one with the most files: chunks, probe candidates, inspections, hashes and counts move; the par2's shared name when none was readable; release_merges keeps what went where. Obfuscated uploads whose files were each a release of their own ("[9/20] - "14f5999a…"", hep: 13,232) become one. Also: corpus_sweep.py counts the marked ranges without psql's command tag. Co-Authored-By: Claude Opus 5.5 --- REBUILD.md | 11 ++ app/services/par2_merger.ts | 154 ++++++++++++++++++ app/services/prober.ts | 28 +++- commands/releases_merge.ts | 33 ++++ .../migrations/1770000002200_par2_merge.ts | 60 +++++++ database/schema.ts | 103 +++++++++++- deploy/README.md | 1 + deploy/compose.yaml | 6 + deploy/corpus_sweep.py | 3 +- deploy/deploy.sh | 4 +- lib/probe/inspect.ts | 18 ++ tests/functional/par2_merger.spec.ts | 65 ++++++++ tests/unit/probe/inspect.spec.ts | 25 ++- 13 files changed, 502 insertions(+), 9 deletions(-) create mode 100644 app/services/par2_merger.ts create mode 100644 commands/releases_merge.ts create mode 100644 database/migrations/1770000002200_par2_merge.ts create mode 100644 tests/functional/par2_merger.spec.ts diff --git a/REBUILD.md b/REBUILD.md index 32915fc..259cfa4 100644 --- a/REBUILD.md +++ b/REBUILD.md @@ -228,6 +228,17 @@ reads the probe queue's first articles (`lone`, `par2`, `archive`) and stores per release whether it's passworded and its inner name, which search, the web interface and Newznab use. +Built: the par2 merge (2026-10-12; `releases:merge`, the `merger` service, +LOG.md 2026-10-12 (13)). The prober stores per probed file start the MD5 of +its first 16 KiB (`file_hashes`) and per par2 body its set's file list +(`par2_files`, `par2_sets`), which holds the same hash per file whatever the +file was renamed to. Every minute the merger merges the releases of one +group, posted within 48 hours of a release holding the set's par2, whose +files the set describes, into the one with the most files (chunks, +candidates, inspections, counts; the par2's shared name when none was +readable); `release_merges` keeps what went where. Obfuscated uploads whose +files were each a release of their own (`lone` candidates) become one. + **Backlog: voyager 1's features not in voyager 2** (CONCEPT.md decisions; voyager 1's data exported to `/mnt/bulk/usenet-rebuild/v1-export/` on homelab: `voyager1-tables.dump` with 2.76 M spots, upload signatures, diff --git a/app/services/par2_merger.ts b/app/services/par2_merger.ts new file mode 100644 index 0000000..2cc174d --- /dev/null +++ b/app/services/par2_merger.ts @@ -0,0 +1,154 @@ +import db from '@adonisjs/lucid/services/db' +import logger from '@adonisjs/core/services/logger' +import { par2ReleaseName } from '#lib/release/index' +import { looksRandom } from '#lib/release/random' + +/** Releases of a set further apart than this (posted) aren't one upload: a repost is merged separately (decision d). */ +const MAX_SPREAD_MS = 48 * 3600 * 1000 +/** Between rounds. */ +const ROUND_MS = 60_000 +/** A round looks back this far past the previous one's start (rows committed while it ran). */ +const OVERLAP_MS = 60_000 + +export interface MergeRound { + sets: number + merges: number + releases: number +} + +interface Member { + id: number + release_key: string + group_id: number + files: number + posted_at: Date | null + readable: boolean +} + +/** + * The par2 merge (REBUILD.md, phase 2): releases whose files a par2 set + * describes are one upload. Obfuscated uploads post each file under a name of + * its own ("[9/20] - "14f5999a7bd4680b0720e524cac0b3b5"", hep: 13,232 + * one-file releases), so the header pass can't group them; the prober reads + * each such file's first article (a `lone` candidate) and stores the MD5 of + * its first 16 KiB, and a par2 body's file list holds the same hash per file, + * whatever the file was renamed to. Releases in one group, posted within + * MAX_SPREAD_MS of a release that holds the set's par2, become the one with + * the most files: its chunks, probe candidates, inspections and hashes, the + * sums of its counts, and when none of them has a readable name the name the + * par2's file names share. Each merge is kept in release_merges. + */ +export class Par2Merger { + private stopped = false + private since = new Date(0) + + stop() { + this.stopped = true + } + + async run() { + while (!this.stopped) { + const started = Date.now() + try { + const round = await this.round() + if (round.merges > 0) logger.info(`Merged ${round.releases} releases into ${round.merges} by par2 set (${round.sets} sets checked).`) + } catch (error: any) { + logger.error(`Par2 merge round failed: ${error.message}`) + } + for (let waited = Date.now() - started; waited < ROUND_MS && !this.stopped; waited += 1000) await new Promise((r) => setTimeout(r, 1000)) + } + } + + /** The sets with something new since the last round: a par2 found, or a hash of one of their files. */ + async round(): Promise { + const { rows: now } = await db.rawQuery(`SELECT now() AS at`) + const since = new Date(this.since.getTime() - OVERLAP_MS) + const { rows } = await db.rawQuery( + `SELECT set_id FROM par2_sets WHERE created_at >= ? + UNION + SELECT f.set_id FROM file_hashes h JOIN par2_files f ON f.hash16k = h.hash16k AND f.length = h.size WHERE h.created_at >= ?`, + [since, since] + ) + const round: MergeRound = { sets: rows.length, merges: 0, releases: 0 } + for (const { set_id: setId } of rows) { + if (this.stopped) return round + const merged = await this.mergeSet(setId.trim()) + if (merged > 0) { + round.merges++ + round.releases += merged + 1 + } + } + this.since = now[0].at + return round + } + + /** Merges the releases of one set; how many went into another. */ + async mergeSet(setId: string): Promise { + return db.transaction(async (trx) => { + const { rows } = await trx.rawQuery( + `WITH m AS ( + SELECT release_id, true AS holds FROM par2_sets WHERE set_id = ? + UNION ALL + SELECT h.release_id, false FROM par2_files f JOIN file_hashes h ON h.hash16k = f.hash16k AND h.size = f.length WHERE f.set_id = ? + ), d AS (SELECT release_id, bool_or(holds) AS holds FROM m GROUP BY 1) + SELECT r.id, r.release_key, r.group_id, r.files, r.posted_at, r.readable, d.holds + FROM releases r JOIN d ON d.release_id = r.id ORDER BY r.id FOR UPDATE OF r`, + [setId, setId] + ) + const anchor = rows.find((r: Member & { holds: boolean }) => r.holds) + if (!anchor) return 0 + const at = (r: Member) => (r.posted_at ? new Date(r.posted_at).getTime() : null) + const near = (r: Member) => at(r) === null || at(anchor) === null || Math.abs(at(r)! - at(anchor)!) <= MAX_SPREAD_MS + const members: Member[] = rows.filter((r: Member) => r.group_id === anchor.group_id && near(r)) + if (members.length < 2) return 0 + const target = members.reduce((a, b) => (Number(b.files) > Number(a.files) ? b : a)) + const others = members.filter((m) => m.id !== target.id).map((m) => Number(m.id)) + const name = members.every((m) => !m.readable) ? await this.setName(trx, setId) : undefined + + await trx.rawQuery( + `INSERT INTO release_merges (from_id, from_key, into_id, set_id, files) + SELECT id, release_key, ?, ?, files FROM releases WHERE id = ANY(?)`, + [target.id, setId, others] + ) + for (const table of ['release_chunks', 'probe_candidates', 'release_inspections', 'file_hashes']) { + await trx.rawQuery(`UPDATE ${table} SET release_id = ? WHERE release_id = ANY(?)`, [target.id, others]) + } + // A set held by the target already isn't held twice. + await trx.rawQuery(`DELETE FROM par2_sets WHERE release_id = ANY(?) AND set_id IN (SELECT set_id FROM par2_sets WHERE release_id = ?)`, [others, target.id]) + await trx.rawQuery(`UPDATE par2_sets SET release_id = ? WHERE release_id = ANY(?)`, [target.id, others]) + await trx.rawQuery( + `UPDATE releases t SET + files = t.files + s.files, segments = t.segments + s.segments, bytes = t.bytes + s.bytes, chunks = t.chunks + s.chunks, + first_article = least(t.first_article, s.first_article), last_article = greatest(t.last_article, s.last_article), + posted_at = least(t.posted_at, s.posted_at), last_posted_at = greatest(t.last_posted_at, s.last_posted_at), + first_seen_at = least(t.first_seen_at, s.first_seen_at), last_seen_at = greatest(t.last_seen_at, s.last_seen_at), + readable = t.readable OR s.readable, has_par2 = t.has_par2 OR s.has_par2, passworded = t.passworded OR s.passworded, + file_total = greatest(t.file_total, s.file_total), inner_name = coalesce(t.inner_name, s.inner_name), + name = coalesce(nullif(?, ''), t.name) + FROM (SELECT sum(files) AS files, sum(segments) AS segments, sum(bytes) AS bytes, sum(chunks) AS chunks, + min(first_article) AS first_article, max(last_article) AS last_article, min(posted_at) AS posted_at, + max(last_posted_at) AS last_posted_at, min(first_seen_at) AS first_seen_at, max(last_seen_at) AS last_seen_at, + bool_or(readable) AS readable, bool_or(has_par2) AS has_par2, bool_or(passworded) AS passworded, + max(file_total) AS file_total, max(inner_name) AS inner_name + FROM releases WHERE id = ANY(?)) s + WHERE t.id = ?`, + [name ?? '', others, target.id] + ) + // Completeness as the settler decides it, once settled. + await trx.rawQuery( + `UPDATE releases SET complete = CASE WHEN file_total IS NULL THEN NULL ELSE files >= file_total END WHERE id = ? AND settled_at IS NOT NULL`, + [target.id] + ) + await trx.rawQuery(`DELETE FROM releases WHERE id = ANY(?)`, [others]) + return others.length + }) + } + + /** The name a set's files share (par2ReleaseName), for an upload none of whose names is readable. */ + private async setName(trx: any, setId: string): Promise { + const { rows } = await trx.rawQuery(`SELECT name FROM par2_files WHERE set_id = ?`, [setId]) + const name = par2ReleaseName(rows.map((r: { name: string }) => r.name)) + // A par2 made after the renaming names the random names again. + return name && !looksRandom(name) ? name.slice(0, 255) : undefined + } +} diff --git a/app/services/prober.ts b/app/services/prober.ts index 45c21d1..e073f56 100644 --- a/app/services/prober.ts +++ b/app/services/prober.ts @@ -159,8 +159,10 @@ export class Prober { `INSERT INTO release_inspections (release_id, candidate_id, kind, result) SELECT v.release_id, v.candidate_id, v.kind, v.result FROM jsonb_to_recordset(?::jsonb) AS v(release_id bigint, candidate_id bigint, kind text, result jsonb) ON CONFLICT (release_id, candidate_id) DO UPDATE SET result = excluded.result, created_at = now()`, - [JSON.stringify(done.map((o) => ({ release_id: o.candidate.releaseId, candidate_id: o.candidate.id, kind: o.candidate.kind, result: o.inspection })))] + // A par2 set's file list goes to par2_files, not into every inspection. + [JSON.stringify(done.map((o) => ({ release_id: o.candidate.releaseId, candidate_id: o.candidate.id, kind: o.candidate.kind, result: { ...o.inspection, par2: undefined } })))] ) + await this.storeForMerge(trx, done) // Each touched release from all its inspections (earlier rounds may have stored others). const ids = [...new Set(done.map((o) => o.candidate.releaseId))] const { rows } = await trx.rawQuery( @@ -187,6 +189,30 @@ export class Prober { }) } + /** What the par2 merge matches (app/services/par2_merger.ts): files' first-16-KiB hashes, par2 sets' file lists. */ + private async storeForMerge(trx: any, done: Outcome[]) { + const hashed = done.filter((o) => o.inspection!.hash16k) + if (hashed.length > 0) { + await trx.rawQuery( + `INSERT INTO file_hashes (candidate_id, release_id, hash16k, size) + SELECT v.candidate_id, v.release_id, v.hash16k, v.size FROM jsonb_to_recordset(?::jsonb) AS v(candidate_id bigint, release_id bigint, hash16k text, size bigint) + ON CONFLICT (candidate_id) DO NOTHING`, + [JSON.stringify(hashed.map((o) => ({ candidate_id: o.candidate.id, release_id: o.candidate.releaseId, hash16k: o.inspection!.hash16k, size: o.inspection!.size })))] + ) + } + for (const o of done) { + const set = o.inspection!.par2 + if (!set) continue + await trx.rawQuery( + `INSERT INTO par2_files (set_id, file_id, hash16k, length, name) + SELECT ?, v.file_id, v.hash16k, v.length, v.name FROM jsonb_to_recordset(?::jsonb) AS v(file_id text, hash16k text, length bigint, name text) + ON CONFLICT DO NOTHING`, + [set.setId, JSON.stringify(set.files.map((f) => ({ file_id: f.fileId, hash16k: f.hash16k, length: f.length, name: f.name })))] + ) + await trx.rawQuery(`INSERT INTO par2_sets (set_id, release_id, files) VALUES (?, ?, ?) ON CONFLICT DO NOTHING`, [set.setId, o.candidate.releaseId, set.files.length]) + } + } + private report() { const seconds = (Date.now() - this.stats.since) / 1000 if (seconds < REPORT_MS / 1000) return diff --git a/commands/releases_merge.ts b/commands/releases_merge.ts new file mode 100644 index 0000000..b8d2468 --- /dev/null +++ b/commands/releases_merge.ts @@ -0,0 +1,33 @@ +import { BaseCommand, flags } from '@adonisjs/core/ace' +import type { CommandOptions } from '@adonisjs/core/types/ace' +import { Par2Merger } from '#services/par2_merger' + +/** The par2 merge (REBUILD.md, phase 2): releases whose files one par2 set describes become one. */ +export default class ReleasesMerge extends BaseCommand { + static commandName = 'releases:merge' + static description = 'Merges the releases whose files one par2 set describes (what the prober read), every minute' + static options: CommandOptions = { startApp: true, staysAlive: true } + + @flags.boolean({ description: 'One round over every set, then exit' }) + declare once: boolean + + async run() { + const merger = new Par2Merger() + if (this.once) { + const round = await merger.round() + this.logger.info(`${round.sets} sets checked: ${round.releases} releases merged into ${round.merges}.`) + await this.terminate() + return + } + let done: () => void + const finished = new Promise((resolve) => (done = resolve)) + this.app.terminating(async () => { + merger.stop() + await finished + }) + await merger.run() + this.logger.info('Merger stopped.') + done!() + await this.terminate() + } +} diff --git a/database/migrations/1770000002200_par2_merge.ts b/database/migrations/1770000002200_par2_merge.ts new file mode 100644 index 0000000..88f9cc0 --- /dev/null +++ b/database/migrations/1770000002200_par2_merge.ts @@ -0,0 +1,60 @@ +import { BaseSchema } from '@adonisjs/lucid/schema' + +/** + * The par2 merge (REBUILD.md, phase 2; app/services/par2_merger.ts): what the + * prober read that ties files to a par2 set, and the merges made from it. + * - file_hashes: per probed article that starts a file, the MD5 of its first 16 KiB and the file's size. + * - par2_files: per par2 recovery set, every file it describes (FileDesc: the same 16 KiB MD5, length, name). + * - par2_sets: which releases hold a par2 file of a set. + * - release_merges: releases merged into another, by set. + */ +export default class extends BaseSchema { + async up() { + this.schema.raw(` + CREATE TABLE file_hashes ( + candidate_id bigint PRIMARY KEY, + release_id bigint NOT NULL, + hash16k char(32) NOT NULL, + size bigint NOT NULL, + created_at timestamptz NOT NULL DEFAULT now() + )`) + this.schema.raw(`CREATE INDEX file_hashes_hash_idx ON file_hashes (hash16k)`) + this.schema.raw(`CREATE INDEX file_hashes_release_idx ON file_hashes (release_id)`) + this.schema.raw(`CREATE INDEX file_hashes_created_idx ON file_hashes (created_at)`) + this.schema.raw(` + CREATE TABLE par2_files ( + set_id char(32) NOT NULL, + file_id char(32) NOT NULL, + hash16k char(32) NOT NULL, + length bigint NOT NULL, + name text NOT NULL, + PRIMARY KEY (set_id, file_id) + )`) + this.schema.raw(`CREATE INDEX par2_files_hash_idx ON par2_files (hash16k)`) + this.schema.raw(` + CREATE TABLE par2_sets ( + set_id char(32) NOT NULL, + release_id bigint NOT NULL, + files integer NOT NULL, + created_at timestamptz NOT NULL DEFAULT now(), + PRIMARY KEY (set_id, release_id) + )`) + this.schema.raw(`CREATE INDEX par2_sets_release_idx ON par2_sets (release_id)`) + this.schema.raw(`CREATE INDEX par2_sets_created_idx ON par2_sets (created_at)`) + this.schema.raw(` + CREATE TABLE release_merges ( + id bigserial PRIMARY KEY, + from_id bigint NOT NULL, + from_key varchar(48) NOT NULL, + into_id bigint NOT NULL, + set_id char(32) NOT NULL, + files integer NOT NULL, + created_at timestamptz NOT NULL DEFAULT now() + )`) + this.schema.raw(`CREATE INDEX release_merges_into_idx ON release_merges (into_id)`) + } + + async down() { + this.schema.raw(`DROP TABLE IF EXISTS file_hashes, par2_files, par2_sets, release_merges`) + } +} diff --git a/database/schema.ts b/database/schema.ts index ddc7964..4380e60 100644 --- a/database/schema.ts +++ b/database/schema.ts @@ -32,6 +32,21 @@ export class AuthAccessTokenSchema extends BaseModel { declare updatedAt: DateTime | null } +export class FileHashSchema extends BaseModel { + static $columns = ['candidateId', 'createdAt', 'hash16K', 'releaseId', 'size'] as const + $columns = FileHashSchema.$columns + @column({ isPrimary: true }) + declare candidateId: bigint | number + @column.dateTime({ autoCreate: true }) + declare createdAt: DateTime + @column() + declare hash16K: string + @column() + declare releaseId: bigint | number + @column() + declare size: bigint | number +} + export class FragmentSchema extends BaseModel { static $columns = ['articleNumber', 'fileKey', 'fragment', 'id', 'rangeId', 'segments'] as const $columns = FragmentSchema.$columns @@ -50,7 +65,7 @@ export class FragmentSchema extends BaseModel { } export class GroupSchema extends BaseModel { - static $columns = ['active', 'createdAt', 'high', 'id', 'low', 'name', 'plannedAt', 'priority'] as const + static $columns = ['active', 'createdAt', 'high', 'id', 'low', 'name', 'plannedAt', 'priority', 'tail'] as const $columns = GroupSchema.$columns @column() declare active: boolean @@ -68,6 +83,8 @@ export class GroupSchema extends BaseModel { declare plannedAt: DateTime | null @column() declare priority: number + @column() + declare tail: boolean } export class NodeAccessTokenSchema extends BaseModel { @@ -108,6 +125,34 @@ export class NodeSchema extends BaseModel { declare name: string } +export class Par2FileSchema extends BaseModel { + static $columns = ['fileId', 'hash16K', 'length', 'name', 'setId'] as const + $columns = Par2FileSchema.$columns + @column() + declare fileId: string + @column() + declare hash16K: string + @column() + declare length: bigint | number + @column() + declare name: string + @column({ isPrimary: true }) + declare setId: string +} + +export class Par2SetSchema extends BaseModel { + static $columns = ['createdAt', 'files', 'releaseId', 'setId'] as const + $columns = Par2SetSchema.$columns + @column.dateTime({ autoCreate: true }) + declare createdAt: DateTime + @column() + declare files: number + @column() + declare releaseId: bigint | number + @column({ isPrimary: true }) + declare setId: string +} + export class ProbeCandidateSchema extends BaseModel { static $columns = ['createdAt', 'file', 'id', 'kind', 'messageId', 'releaseId', 'status'] as const $columns = ProbeCandidateSchema.$columns @@ -128,10 +173,12 @@ export class ProbeCandidateSchema extends BaseModel { } export class RangeSchema extends BaseModel { - static $columns = ['attempts', 'doneAt', 'endId', 'error', 'groupId', 'headers', 'heartbeatAt', 'id', 'leasedAt', 'nextId', 'nodeId', 'priority', 'ruleVersion', 'spooledAt', 'spooledChunks', 'startId', 'status'] as const + static $columns = ['attempts', 'capture', 'doneAt', 'endId', 'error', 'groupId', 'headers', 'heartbeatAt', 'id', 'leasedAt', 'nextId', 'nodeId', 'plainChunks', 'priority', 'ruleVersion', 'spooledAt', 'spooledChunks', 'startId', 'status'] as const $columns = RangeSchema.$columns @column() declare attempts: number + @column() + declare capture: boolean @column.dateTime() declare doneAt: DateTime | null @column() @@ -153,6 +200,8 @@ export class RangeSchema extends BaseModel { @column() declare nodeId: number | null @column() + declare plainChunks: any + @column() declare priority: number @column() declare ruleVersion: number | null @@ -187,12 +236,48 @@ export class ReleaseChunkSchema extends BaseModel { declare segments: number } +export class ReleaseInspectionSchema extends BaseModel { + static $columns = ['candidateId', 'createdAt', 'kind', 'releaseId', 'result'] as const + $columns = ReleaseInspectionSchema.$columns + @column() + declare candidateId: bigint | number + @column.dateTime({ autoCreate: true }) + declare createdAt: DateTime + @column() + declare kind: string + @column({ isPrimary: true }) + declare releaseId: bigint | number + @column() + declare result: any +} + +export class ReleaseMergeSchema extends BaseModel { + static $columns = ['createdAt', 'files', 'fromId', 'fromKey', 'id', 'intoId', 'setId'] as const + $columns = ReleaseMergeSchema.$columns + @column.dateTime({ autoCreate: true }) + declare createdAt: DateTime + @column() + declare files: number + @column() + declare fromId: bigint | number + @column() + declare fromKey: string + @column({ isPrimary: true }) + declare id: bigint | number + @column() + declare intoId: bigint | number + @column() + declare setId: string +} + export class ReleaseSchema extends BaseModel { - static $columns = ['bytes', 'chunks', 'complete', 'fileTotal', 'files', 'firstArticle', 'firstSeenAt', 'groups', 'hasPar2', 'id', 'lastArticle', 'lastSeenAt', 'name', 'postedAt', 'poster', 'readable', 'releaseKey', 'segments', 'settledAt', 'source'] as const + static $columns = ['bytes', 'category', 'chunks', 'complete', 'fileTotal', 'files', 'firstArticle', 'firstSeenAt', 'groupId', 'groups', 'hasPar2', 'id', 'identityKey', 'innerName', 'inspectedAt', 'lastArticle', 'lastPostedAt', 'lastSeenAt', 'name', 'passworded', 'postedAt', 'poster', 'readable', 'releaseKey', 'segments', 'settledAt', 'source'] as const $columns = ReleaseSchema.$columns @column() declare bytes: bigint | number @column() + declare category: number | null + @column() declare chunks: number @column() declare complete: boolean | null @@ -205,17 +290,29 @@ export class ReleaseSchema extends BaseModel { @column.dateTime() declare firstSeenAt: DateTime @column() + declare groupId: number | null + @column() declare groups: any @column() declare hasPar2: boolean @column({ isPrimary: true }) declare id: bigint | number @column() + declare identityKey: string + @column() + declare innerName: string | null + @column.dateTime() + declare inspectedAt: DateTime | null + @column() declare lastArticle: bigint | number | null @column.dateTime() + declare lastPostedAt: DateTime | null + @column.dateTime() declare lastSeenAt: DateTime @column() declare name: string + @column() + declare passworded: boolean | null @column.dateTime() declare postedAt: DateTime | null @column() diff --git a/deploy/README.md b/deploy/README.md index 1af6096..966dc5a 100644 --- a/deploy/README.md +++ b/deploy/README.md @@ -19,6 +19,7 @@ homelab with the deploy key, whose forced command | `node-api` | `usenet-rebuild` | the edge workers' API (leases, streams, batches); port 3335 on the compose network only (`HTTP_ROUTES=node`) | | `coordinator` | `usenet-rebuild` | prefetching (all NNTP connections), planning, the tail, merging, settling | | `prober` | `usenet-rebuild` | phase 2: reads the probe queue's articles over `PROBE_CONNECTIONS` (6) connections, stores what they show | +| `merger` | `usenet-rebuild` | the par2 merge: releases whose files one par2 set describes become one (no NNTP) | | `edge` | `usenet-rebuild-edge` (target `edge`) | the Rust worker, `--threads $EDGE_THREADS` (8) | | `edge-ts` | `usenet-rebuild` | the TypeScript worker, profile `edge-ts`, not started | | `ace` | `usenet-rebuild` | one-off ace commands, profile `tools` | diff --git a/deploy/compose.yaml b/deploy/compose.yaml index 2f59289..7db031c 100644 --- a/deploy/compose.yaml +++ b/deploy/compose.yaml @@ -108,6 +108,12 @@ services: <<: *app-env PROBE_CONNECTIONS: ${PROBE_CONNECTIONS:-6} + # The par2 merge: releases whose files one par2 set describes (what the prober read) become one. No NNTP. + merger: + <<: *app + command: ['node', 'ace', 'releases:merge'] + volumes: [] + # The Rust worker; it has no NNTP account and talks to node-api only. # Stopped before node-api (a worker holding a stream keeps the server from # exiting). diff --git a/deploy/corpus_sweep.py b/deploy/corpus_sweep.py index cff37a2..f9b9d86 100644 --- a/deploy/corpus_sweep.py +++ b/deploy/corpus_sweep.py @@ -42,7 +42,8 @@ def psql(sql): ['docker', 'compose', 'exec', '-T', 'postgres', 'psql', '-U', env('DB_USER'), '-d', env('DB_DATABASE'), '-At', '-F', '\t', '-v', 'ON_ERROR_STOP=1'], input=sql, capture_output=True, text=True, check=True, ).stdout - return [line.split('\t') for line in out.splitlines() if line] + # Without the command tag a RETURNING prints after its rows ("UPDATE 100"). + return [line.split('\t') for line in out.splitlines() if line and not line.startswith(('UPDATE ', 'INSERT ', 'DELETE '))] def corpus_bytes(): diff --git a/deploy/deploy.sh b/deploy/deploy.sh index c680f82..6ab05be 100755 --- a/deploy/deploy.sh +++ b/deploy/deploy.sh @@ -45,10 +45,10 @@ docker compose run --rm --no-deps migrate # stream keeps the node API from exiting. echo "== recreating app services" docker compose stop edge -docker compose up -d --no-deps node-api web coordinator prober +docker compose up -d --no-deps node-api web coordinator prober merger docker compose up -d --no-deps edge echo "$SHA $(date -Is)" >> "$ROOT/deployed.log" # The images this replaced (321 MB each; the root filesystem is the NVMe, with a 10 GB floor). docker image prune -f --filter label=com.docker.compose.project=usenet-rebuild >/dev/null -docker compose ps --format '{{.Service}} {{.Status}}' node-api web coordinator prober edge +docker compose ps --format '{{.Service}} {{.Status}}' node-api web coordinator prober merger edge echo "== deployed $SHA" diff --git a/lib/probe/inspect.ts b/lib/probe/inspect.ts index 4f5384f..05a702a 100644 --- a/lib/probe/inspect.ts +++ b/lib/probe/inspect.ts @@ -1,3 +1,4 @@ +import { createHash } from 'node:crypto' import { decodeSegment, YencCrcError, YencError, type DecodedSegment } from '#lib/yenc/index' import { isPar2, parsePar2 } from '#lib/par2/index' import { parseRarHeaders } from '#lib/rar/index' @@ -16,8 +17,17 @@ export interface Inspection { entries?: number /** An archive with encrypted entries or headers. */ passworded?: boolean + /** + * MD5 of the file's first 16 KiB (hex), when this article starts the file: what a par2 set's FileDesc + * holds per file, so a renamed file is recognised by it (the par2 merge; app/services/par2_merger.ts). + */ + 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 }[] } } +const HASH_BYTES = 16 * 1024 + /** Names kept per inspection: the catalog wants the content's name, not every volume's. */ export const MAX_NAMES = 8 const MAX_NAME_LENGTH = 200 @@ -46,12 +56,20 @@ function largestFirst(items: { name: string; size: number }[]): string[] { export function inspect(body: Buffer): 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) } + // 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)) { + result.hash16k = createHash('md5').update(segment.data.subarray(0, HASH_BYTES)).digest('hex') + } if (isPar2(segment.data)) { // An index file fits one article; what a larger one cut off is skipped by the parser. const set = parsePar2(segment.data) if (set) { result.entries = set.files.length 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) })), + } } return result } diff --git a/tests/functional/par2_merger.spec.ts b/tests/functional/par2_merger.spec.ts new file mode 100644 index 0000000..c24a2b6 --- /dev/null +++ b/tests/functional/par2_merger.spec.ts @@ -0,0 +1,65 @@ +import { test } from '@japa/runner' +import db from '@adonisjs/lucid/services/db' +import testUtils from '@adonisjs/core/services/test_utils' +import { Par2Merger } from '#services/par2_merger' + +const hash = (c: string) => c.repeat(32) + +test.group('Par2Merger', (group) => { + group.each.setup(() => testUtils.db().withGlobalTransaction()) + + test('merges the one-file releases of an obfuscated upload by its par2 set, named by the par2', async ({ assert }) => { + const at = new Date('2022-07-08T03:46:00Z') + const release = async (key: string, posted: Date, groupId: number, files = 1) => { + const [row] = await db + .table('releases') + .insert({ + release_key: key.padEnd(40, '0'), identity_key: key.padEnd(40, '0'), name: key, source: 'stem', groups: '[]', group_id: groupId, + files, segments: files * 10, bytes: files * 1000, first_article: 100, last_article: 200, posted_at: posted, readable: false, file_total: 20, + settled_at: posted, + }) + .returning('id') + return Number(row.id) + } + // The par2 index, two data files of the upload, the same file reposted a week later, and one in another group. + const par2 = await release('14f5999a', at, 1) + const a = await release('2b7c00e1', new Date(at.getTime() + 60_000), 1) + const b = await release('9d01aa3f', new Date(at.getTime() + 120_000), 1, 2) + const later = await release('5e5e5e5e', new Date(at.getTime() + 7 * 86400_000), 1) + const elsewhere = await release('6f6f6f6f', at, 2) + let candidate = 1000 + const hashed = async (releaseId: number, h: string, size: number) => { + candidate++ + await db.table('probe_candidates').insert({ id: candidate, release_id: releaseId, kind: 'lone', message_id: `<${candidate}@x>`, file: 'x', status: 'done' }) + await db.table('file_hashes').insert({ candidate_id: candidate, release_id: releaseId, hash16k: h, size }) + } + await hashed(a, hash('a'), 104857600) + await hashed(b, hash('b'), 104857600) + await hashed(later, hash('a'), 104857600) + await hashed(elsewhere, hash('b'), 104857600) + // A different size: not the same file. + await hashed(await release('7a7a7a7a', at, 1), hash('a'), 5) + const setId = 'c'.repeat(32) + await db.table('par2_files').multiInsert([ + { set_id: setId, file_id: 'd'.repeat(32), hash16k: hash('a'), length: 104857600, name: 'Some.Movie.2022.1080p.part01.rar' }, + { set_id: setId, file_id: 'e'.repeat(32), hash16k: hash('b'), length: 104857600, name: 'Some.Movie.2022.1080p.part02.rar' }, + ]) + await db.table('par2_sets').insert({ set_id: setId, release_id: par2, files: 2 }) + + const round = await new Par2Merger().round() + assert.deepEqual(round, { sets: 1, merges: 1, releases: 3 }) + const target = await db.from('releases').where('id', b).firstOrFail() + assert.equal(target.files, 4) + assert.equal(target.segments, 40) + assert.equal(target.name, 'Some.Movie.2022.1080p') + assert.isTrue(target.complete === false) + assert.deepEqual((await db.from('releases').whereIn('id', [par2, a]).select('id')).length, 0) + assert.equal((await db.from('releases').whereIn('id', [later, elsewhere]).select('id')).length, 2) + assert.deepEqual((await db.from('probe_candidates').where('release_id', b).orderBy('id').select('id')).map((r) => Number(r.id)), [1001, 1002]) + assert.equal((await db.from('par2_sets').where('set_id', setId).firstOrFail()).release_id, String(b)) + assert.equal((await db.from('release_merges').where('into_id', b).count('* as n'))[0].n, '2') + + // Again: nothing new. + assert.deepEqual(await new Par2Merger().round().then((r) => r.merges), 0) + }) +}) diff --git a/tests/unit/probe/inspect.spec.ts b/tests/unit/probe/inspect.spec.ts index 0b97080..c1bd1ae 100644 --- a/tests/unit/probe/inspect.spec.ts +++ b/tests/unit/probe/inspect.spec.ts @@ -80,7 +80,28 @@ test.group('probe / inspect', () => { const setId = Buffer.alloc(16, 3) const par2 = Buffer.concat([fileDesc(setId, 'Movie.2020.part02.rar', 100), fileDesc(setId, 'Movie.2020.part01.rar', 200)]) const result = inspect(article(par2, 'a8f3k2.par2')) - assert.deepEqual(result, { yencName: 'a8f3k2.par2', size: par2.length, type: 'par2', entries: 2, names: ['Movie.2020.part01.rar', 'Movie.2020.part02.rar'] }) + const { par2: set, hash16k, ...rest } = result + assert.deepEqual(rest, { yencName: 'a8f3k2.par2', size: par2.length, type: 'par2', entries: 2, names: ['Movie.2020.part01.rar', 'Movie.2020.part02.rar'] }) + // The whole set for the par2 merge: per file the hash of its first 16 KiB, its length and name. + assert.equal(set!.setId, setId.toString('hex')) + assert.deepEqual( + set!.files.map((f) => [f.name, f.length, f.hash16k]), + [ + ['Movie.2020.part02.rar', 100, md5(Buffer.from('Movie.2020.part02.rar-16k')).toString('hex')], + ['Movie.2020.part01.rar', 200, md5(Buffer.from('Movie.2020.part01.rar-16k')).toString('hex')], + ] + ) + assert.equal(hash16k, md5(par2).toString('hex')) + }) + + test("the hash of a file's first 16 KiB, when the article starts the file", ({ assert }) => { + const data = Buffer.alloc(40_000, 7) + assert.equal(inspect(article(data, 'x')).hash16k, md5(data.subarray(0, 16384)).toString('hex')) + const part = Buffer.from( + [`=ybegin part=2 total=3 line=128 size=120000 name=x`, `=ypart begin=40001 end=80000`, yencode.encode(data).toString('latin1'), `=yend size=40000 part=2 pcrc32=${crc32Hex(data)}`].join('\r\n'), + 'latin1' + ) + assert.isUndefined(inspect(part).hash16k) }) test('a RAR volume: its entries, and whether they are encrypted', ({ assert }) => { @@ -96,7 +117,7 @@ test.group('probe / inspect', () => { test('other data: the yEnc name and what the bytes are', ({ assert }) => { const mkv = Buffer.concat([Buffer.from([0x1a, 0x45, 0xdf, 0xa3]), Buffer.alloc(100)]) - assert.deepEqual(inspect(article(mkv, 'Real.Name.mkv')), { yencName: 'Real.Name.mkv', size: 104, type: 'matroska' }) + assert.deepEqual(inspect(article(mkv, 'Real.Name.mkv')), { yencName: 'Real.Name.mkv', size: 104, type: 'matroska', hash16k: md5(mkv).toString('hex') }) }) test('not yEnc: throws', ({ assert }) => {