From ebe9f8bb36485258c2167df7d6cc383950c8df5e Mon Sep 17 00:00:00 2001 From: Daan Meijer Date: Fri, 23 Oct 2026 12:00:19 +0200 Subject: [PATCH] Decide completeness by par2 set for releases without a counter 59% of settled releases had complete = NULL: without an [i/N] counter the settler left it unknown, and the "phase 2 reads par2 sets for those" step was never built. A release is now complete when every file its par2 sets describe is there by name or by its first-16-KiB hash, or, for an upload renamed after its par2 was made, when it has as many non-par2 files and bytes. Decided when a par2 index is probed, after a par2 merge, and before the pruner drops a set's file list; the batch store, settler and merger keep a decided value instead of resetting it to NULL. releases:par2-complete decides the releases probed before this. Co-Authored-By: Claude Opus 5.5 --- REBUILD.md | 12 ++-- app/services/batch_store.ts | 5 +- app/services/nzb_service.ts | 4 +- app/services/par2_completeness.ts | 53 +++++++++++++++++ app/services/par2_merger.ts | 6 +- app/services/par2_pruner.ts | 6 ++ app/services/prober.ts | 3 + app/services/settler.ts | 5 +- commands/releases_par2_complete.ts | 38 ++++++++++++ tests/functional/par2_completeness.spec.ts | 67 ++++++++++++++++++++++ 10 files changed, 187 insertions(+), 12 deletions(-) create mode 100644 app/services/par2_completeness.ts create mode 100644 commands/releases_par2_complete.ts create mode 100644 tests/functional/par2_completeness.spec.ts diff --git a/REBUILD.md b/REBUILD.md index 8a2213d..65640c7 100644 --- a/REBUILD.md +++ b/REBUILD.md @@ -189,10 +189,14 @@ dropping garbage; 28 TB of compressed overview traffic. Throughput goal: number; merged when the neighbouring range is done, then written as a chunk. Never staged per segment. -8. **Settling**. A release is settled when the ranges covering its posting - span are done and no chunk arrived for a day: completeness is computed - then (expected vs found files and segments). Phase 2 picks its work from - settled releases. +8. **Settling**. A release is settled when nothing was added to it for 6 + hours (30 minutes once its last post is that old: history), and its + completeness decided: complete when every file its [i/N] counter + announced was found. Without a counter, by its par2 sets once their index + is probed (`par2_completeness`): every file a set describes is there by + name or by its first 16 KiB, or, renamed, as many files and bytes; decided + again after a par2 merge and before the pruner drops a set's file list. + Segments aren't compared. Phase 2 picks its work from settled releases. 9. **NZB and API**. NZB built from a release's chunks; a minimal Newznab search (names) and get. Web UI later. diff --git a/app/services/batch_store.ts b/app/services/batch_store.ts index cb490f0..ef6c179 100644 --- a/app/services/batch_store.ts +++ b/app/services/batch_store.ts @@ -237,8 +237,9 @@ class BatchStore { last_article = greatest(r.last_article, a.last_article), readable = r.readable OR a.readable, has_par2 = r.has_par2 OR a.has_par2, - -- As it stands (the counter's files all found); settling makes it final. - complete = CASE WHEN coalesce(r.file_total, a.file_total) IS NULL THEN NULL ELSE r.files + a.files >= coalesce(r.file_total, a.file_total) END, + -- As it stands (the counter's files all found); settling makes it final. Without a counter, what its par2 + -- set decided stays (app/services/par2_completeness.ts). + complete = CASE WHEN coalesce(r.file_total, a.file_total) IS NULL THEN r.complete ELSE r.files + a.files >= coalesce(r.file_total, a.file_total) END, last_seen_at = now() FROM a WHERE r.id = a.id`, [JSON.stringify(updates)] diff --git a/app/services/nzb_service.ts b/app/services/nzb_service.ts index 450ad19..1d547a5 100644 --- a/app/services/nzb_service.ts +++ b/app/services/nzb_service.ts @@ -21,8 +21,8 @@ class NzbService { return rows.sort((a, b) => rank(a.source) - rank(b.source)).map((r) => ({ password: r.password, source: r.source })) } - async files(releaseId: number): Promise { - const chunks = await db.from('release_chunks').where('release_id', releaseId).orderBy('id').select('payload') + async files(releaseId: number, client: any = db): Promise { + const chunks = await client.from('release_chunks').where('release_id', releaseId).orderBy('id').select('payload') const files = new Map() for (const chunk of chunks) { for (const file of decodeChunk(chunk.payload)) files.set(`${file.name}\n${file.segments[0]?.[1] ?? ''}`, file) diff --git a/app/services/par2_completeness.ts b/app/services/par2_completeness.ts new file mode 100644 index 0000000..4ab9f0f --- /dev/null +++ b/app/services/par2_completeness.ts @@ -0,0 +1,53 @@ +import NzbService from '#services/nzb_service' + +/** A file a par2 set describes (par2_files). */ +export interface SetFile { + name: string + length: number + hash16k: string +} + +const base = (name: string) => name.replace(/^.*[\\/]/, '').toLowerCase() + +/** + * Whether a release holds every file its par2 sets describe (REBUILD.md, part 8): for one without an [i/N] counter, + * whose completeness the settler leaves unknown. A described file is there when a file of the release has its name, + * or a probed file of it its first 16 KiB and size (file_hashes). An upload renamed after its par2 was made (no + * described name among its files) is complete when it has as many files besides par2 as the set describes, and as + * many bytes (its articles' bytes, a little over the files' with yEnc). + */ +export function par2Complete(set: SetFile[], files: { name: string; size: number }[], hashes: { hash16k: string; size: number }[]): boolean { + const names = new Set(files.map((f) => base(f.name))) + const hashed = new Set(hashes.map((h) => `${h.hash16k}:${h.size}`)) + const byName = set.filter((f) => names.has(base(f.name))).length + if (set.every((f) => names.has(base(f.name)) || hashed.has(`${f.hash16k}:${f.length}`))) return true + if (byName > 0) return false + const data = files.filter((f) => !/\.par2$/i.test(f.name)) + return data.length >= set.length && data.reduce((n, f) => n + f.size, 0) >= set.reduce((n, f) => n + f.length, 0) +} + +/** + * Decides `complete` for the settled releases among `releaseIds` that have no counter and a par2 set with its file + * list (par2_files; the pruner drops it once the set is done, so it decides before). How many it decided. `client`: + * a transaction that moved chunks (a merge) reads them as moved. + */ +export async function decidePar2Completeness(client: any, releaseIds: (number | string)[]): Promise { + if (releaseIds.length === 0) return 0 + const { rows: releases } = await client.rawQuery( + `SELECT id FROM releases WHERE id = ANY(?) AND file_total IS NULL AND settled_at IS NOT NULL`, + [releaseIds.map(String)] + ) + let decided = 0 + for (const { id } of releases) { + const { rows: set } = await client.rawQuery( + `SELECT DISTINCT f.file_id, f.name, f.length::float8 AS length, f.hash16k FROM par2_sets s JOIN par2_files f ON f.set_id = s.set_id WHERE s.release_id = ?`, + [id] + ) + if (set.length === 0) continue + const { rows: hashes } = await client.rawQuery(`SELECT hash16k, size::float8 AS size FROM file_hashes WHERE release_id = ?`, [id]) + const files = (await NzbService.files(Number(id), client)).map((f) => ({ name: f.name, size: f.segments.reduce((n, [, , bytes]) => n + bytes, 0) })) + await client.rawQuery(`UPDATE releases SET complete = ? WHERE id = ?`, [par2Complete(set, files, hashes), id]) + decided++ + } + return decided +} diff --git a/app/services/par2_merger.ts b/app/services/par2_merger.ts index 097a701..5567c74 100644 --- a/app/services/par2_merger.ts +++ b/app/services/par2_merger.ts @@ -5,6 +5,7 @@ import { looksRandom } from '#lib/release/random' import { SidecarFolder } from '#services/sidecar_folder' import { Par2Pruner } from '#services/par2_pruner' import { pruneSevenZipTails } from '#services/seven_zip_tails' +import { decidePar2Completeness } from '#services/par2_completeness' /** 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 @@ -371,12 +372,13 @@ export class Par2Merger { WHERE t.id = ?`, [name ?? '', others, target.id] ) - // Completeness as the settler decides it, once settled. + // Completeness as the settler decides it, once settled; without a counter, by the par2 sets it now holds. 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`, + `UPDATE releases SET complete = CASE WHEN file_total IS NULL THEN complete 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]) + await decidePar2Completeness(trx, [target.id]) return others.length } } diff --git a/app/services/par2_pruner.ts b/app/services/par2_pruner.ts index 4d839ef..e440553 100644 --- a/app/services/par2_pruner.ts +++ b/app/services/par2_pruner.ts @@ -1,4 +1,5 @@ import db from '@adonisjs/lucid/services/db' +import { decidePar2Completeness } from '#services/par2_completeness' /** A set this long past its last holder's par2 has had its merges: one posted later is a repost (decision d). */ const SETTLED_MS = 2 * 86400_000 @@ -86,6 +87,11 @@ export class Par2Pruner { for (const reason of ['complete', 'judged', 'final'] as const) { const ids = done(reason) round[reason] += ids.length + // The holders' completeness while the file lists are there (app/services/par2_completeness.ts). + if (reason !== 'judged' && ids.length > 0) { + const { rows: holders } = await db.rawQuery(`SELECT DISTINCT release_id FROM par2_sets WHERE set_id = ANY(?::char(32)[])`, [ids]) + await decidePar2Completeness(db, holders.map((h: { release_id: string }) => h.release_id)) + } round.rows += await this.drop(ids) } // Past the last set: the next batch starts the next pass. diff --git a/app/services/prober.ts b/app/services/prober.ts index 9911d8a..245c278 100644 --- a/app/services/prober.ts +++ b/app/services/prober.ts @@ -8,6 +8,7 @@ import { YencError } from '#lib/yenc/index' import { inspect, innerName, PROBE_VERSION, type Inspection } from '#lib/probe/index' import { NntpShare } from '#services/nntp_shares' import { DiskSpace } from '#services/disk_space' +import { decidePar2Completeness } from '#services/par2_completeness' interface Candidate { id: number @@ -306,6 +307,8 @@ export class Prober { [jsonb(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) + // A par2 index read: the release's completeness by its set, when it has no counter. + await decidePar2Completeness(trx, [...new Set(done.filter((o) => o.inspection!.par2).map((o) => o.candidate.releaseId))]) // 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( diff --git a/app/services/settler.ts b/app/services/settler.ts index 3bc78e2..38fcf79 100644 --- a/app/services/settler.ts +++ b/app/services/settler.ts @@ -15,7 +15,8 @@ const BATCH = 50_000 * Settles releases (REBUILD.md, part 8): once nothing has been added for * QUIET_HOURS (QUIET_HISTORY_MINUTES once its last post is that old), its completeness is decided: complete when every file the * [i/N] counter announced was found, unknown without a counter (phase 2 - * reads par2 sets for those). Phase 2 picks its work from settled releases. + * decides those by their par2 sets: app/services/par2_completeness.ts). + * Phase 2 picks its work from settled releases. */ class Settler { /** Up to `batches` batches of BATCH (all there are without). */ @@ -25,7 +26,7 @@ class Settler { // Their probe candidates become claimable (Prober.claim), the unreadable ones first. const { rows } = await db.rawQuery( `WITH settled AS ( - UPDATE releases SET settled_at = now(), complete = CASE WHEN file_total IS NULL THEN NULL ELSE files >= file_total END + UPDATE releases SET settled_at = now(), complete = CASE WHEN file_total IS NULL THEN complete ELSE files >= file_total END WHERE id IN (SELECT id FROM releases WHERE settled_at IS NULL AND last_seen_at < now() - make_interval(mins => ?) AND (last_seen_at < now() - make_interval(hours => ?) OR coalesce(last_posted_at, posted_at) < now() - make_interval(hours => ?)) LIMIT ?) RETURNING id, readable, files, bytes, poster, sidecar_only), diff --git a/commands/releases_par2_complete.ts b/commands/releases_par2_complete.ts new file mode 100644 index 0000000..387f970 --- /dev/null +++ b/commands/releases_par2_complete.ts @@ -0,0 +1,38 @@ +import { BaseCommand } from '@adonisjs/core/ace' +import type { CommandOptions } from '@adonisjs/core/types/ace' +import db from '@adonisjs/lucid/services/db' +import { decidePar2Completeness } from '#services/par2_completeness' + +/** Releases per batch. */ +const BATCH = 1000 + +/** + * Decides completeness by par2 set (app/services/par2_completeness.ts) for the releases that had their par2 read + * before it was built (2026-10-23): every holder of a set whose file list is still there, by release id. Once. + */ +export default class ReleasesPar2Complete extends BaseCommand { + static commandName = 'releases:par2-complete' + static description = 'Decides completeness by their par2 sets for settled releases without a counter, then exits' + static options: CommandOptions = { startApp: true } + + async run() { + let after = 0 + let looked = 0 + let decided = 0 + for (;;) { + const { rows } = await db.rawQuery( + `SELECT DISTINCT s.release_id FROM par2_sets s + WHERE s.release_id > ? AND EXISTS (SELECT 1 FROM par2_files f WHERE f.set_id = s.set_id) + ORDER BY s.release_id LIMIT ?`, + [after, BATCH] + ) + if (rows.length === 0) break + const ids = rows.map((r: { release_id: string }) => Number(r.release_id)) + decided += await decidePar2Completeness(db, ids) + looked += ids.length + after = ids.at(-1)! + if (looked % 50_000 < BATCH) this.logger.info(`${looked} releases with a par2 set looked at, ${decided} decided (up to id ${after}).`) + } + this.logger.info(`Done: ${looked} releases with a par2 set looked at, ${decided} decided.`) + } +} diff --git a/tests/functional/par2_completeness.spec.ts b/tests/functional/par2_completeness.spec.ts new file mode 100644 index 0000000..96957aa --- /dev/null +++ b/tests/functional/par2_completeness.spec.ts @@ -0,0 +1,67 @@ +import { test } from '@japa/runner' +import db from '@adonisjs/lucid/services/db' +import testUtils from '@adonisjs/core/services/test_utils' +import BatchStore from '#services/batch_store' +import { decidePar2Completeness, par2Complete } from '#services/par2_completeness' +import { RangeIndexer } from '#lib/edge/index' +import { DEFAULT_RULES } from '#lib/rules/index' +import type { Overview } from '#lib/nntp/index' + +const GROUP = 'alt.binaries.test' + +/** A file posted without an [i/N] counter: the release gets no file_total. */ +function file(name: string, parts: number, at: number): Overview[] { + return Array.from({ length: parts }, (_, i) => [at + i, { + 'subject': `"${name}" yEnc (${i + 1}/${parts})`, 'from': 'p ', 'date': 'Mon, 05 Oct 2026 10:00:00 +0000', + 'message-id': `<${name}-${i + 1}@test>`, ':bytes': '700000', 'xref': `news.example ${GROUP}:${at + i}`, + }] as Overview) +} + +async function store(start: number, overviews: Overview[]) { + const g = (await db.from('groups').where('name', GROUP).first()) ?? (await db.table('groups').insert({ name: GROUP, priority: 1 }).returning('*'))[0] + const [n] = await db.table('nodes').insert({ name: `n${start}` }).returning('id') + const [r] = await db.table('ranges').insert({ group_id: g.id, start_id: start, end_id: start + 999, next_id: start, status: 'leased', node_id: n.id }).returning('id') + const indexer = new RangeIndexer({ group: GROUP, start, end: start + 999, rules: { ...DEFAULT_RULES, samplePercent: 0, droppedSignatures: [], priorities: { [GROUP]: 1 } } }) + indexer.add(overviews, start + 1000) + indexer.finish() + await BatchStore.store(Number(r.id), Number(n.id), indexer.take()) +} + +const setFile = (name: string, length: number, hash16k = name.padEnd(32, '0').slice(0, 32)) => ({ name, length, hash16k }) + +test.group('par2 completeness', (group) => { + group.each.setup(() => testUtils.db().withGlobalTransaction()) + + test('every described file by name or by hash; a renamed upload by its files and bytes', ({ assert }) => { + const set = [setFile('a.mkv', 1000), setFile('a.nfo', 10)] + const files = [{ name: 'a.mkv', size: 1100 }, { name: 'a.nfo', size: 12 }, { name: 'a.par2', size: 50 }] + assert.isTrue(par2Complete(set, files, [])) + assert.isFalse(par2Complete(set, files.slice(0, 1), [])) + // The nfo renamed, but its probe hashed it. + assert.isTrue(par2Complete(set, [files[0], { name: 'x.nfo', size: 12 }], [{ hash16k: set[1].hash16k, size: 10 }])) + // Renamed after the par2 was made (random names): two data files and enough bytes; one missing, or too small. + assert.isTrue(par2Complete(set, [{ name: 'Qx81.bin', size: 1100 }, { name: 'Zr02.bin', size: 12 }, { name: 'Qx.par2', size: 50 }], [])) + assert.isFalse(par2Complete(set, [{ name: 'Qx81.bin', size: 1100 }, { name: 'Qx.par2', size: 50 }], [])) + assert.isFalse(par2Complete(set, [{ name: 'Qx81.bin', size: 500 }, { name: 'Zr02.bin', size: 12 }], [])) + }) + + test('decided once settled with a par2 set, and kept when more files come', async ({ assert }) => { + await store(1, [...file('Some.Upload.mkv', 3, 10), ...file('Some.Upload.par2', 1, 20)]) + const release = await db.from('releases').firstOrFail() + assert.isNull(release.file_total) + await db.table('par2_files').multiInsert([setFile('Some.Upload.mkv', 2_000_000), setFile('Some.Upload.nfo', 100)].map((f, i) => ({ set_id: 's'.repeat(32), file_id: String(i).repeat(32), ...f }))) + await db.table('par2_sets').insert({ set_id: 's'.repeat(32), release_id: release.id, files: 2 }) + // Not settled yet: left alone. + assert.equal(await decidePar2Completeness(db, [release.id]), 0) + await db.from('releases').where('id', release.id).update({ settled_at: new Date() }) + assert.equal(await decidePar2Completeness(db, [release.id]), 1) + assert.isFalse((await db.from('releases').where('id', release.id).firstOrFail()).complete) + // The nfo comes in a later range: complete, and the batch store's own update keeps it. + await store(1001, file('Some.Upload.nfo', 1, 1010)) + await decidePar2Completeness(db, [release.id]) + await store(2001, file('Some.Upload.sfv', 1, 2010)) + const after = await db.from('releases').where('id', release.id).firstOrFail() + assert.equal(after.files, 4) + assert.isTrue(after.complete) + }) +})