diff --git a/REBUILD.md b/REBUILD.md index b40573e..e3c1a0f 100644 --- a/REBUILD.md +++ b/REBUILD.md @@ -242,6 +242,10 @@ It reads what's new an hour (`created_at`) at a time, its place kept after each, and merges only the sets held or matched by two releases or more; a file whose first 16 KiB and size more than 20 files share is a common one and matches nothing (LOG.md 2026-10-20 (85)). +A set's file list is dropped once it can't merge anything more: complete (all +its files in its holder), judged (every holder junk), or final (every holder +inside a run of done ranges, those ranges' candidates probed, its hashes seen +by the merger); `par2_sets` stays (LOG.md 2026-10-20 (87)). **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 diff --git a/app/services/batch_store.ts b/app/services/batch_store.ts index a832c9b..de0cae5 100644 --- a/app/services/batch_store.ts +++ b/app/services/batch_store.ts @@ -316,6 +316,15 @@ class BatchStore { .ignore() .returning('id') out.candidates += added.length + // Files added to a settled release (a later range of its upload): the settler made its candidates claimable + // already and won't again (93,642 waited since 2026-10-13; LOG.md 2026-10-20 (87)). + if (added.length > 0) { + await trx.rawQuery( + `UPDATE probe_candidates pc SET ready = true, unreadable = NOT r.readable FROM releases r + WHERE pc.id = ANY(?) AND r.id = pc.release_id AND r.settled_at IS NOT NULL`, + [added.map((a: { id: string | number }) => String(a.id))] + ) + } } return out diff --git a/app/services/par2_merger.ts b/app/services/par2_merger.ts index 5706433..b1bfa83 100644 --- a/app/services/par2_merger.ts +++ b/app/services/par2_merger.ts @@ -115,7 +115,7 @@ export class Par2Merger { if (!this.stopped && Date.now() - prunedAt >= PRUNE_EVERY_MS) { prunedAt = Date.now() const pruned = await new Par2Pruner().round(PRUNE_BUDGET_MS) - if (pruned.rows) logger.info(`Pruned ${pruned.rows} par2 file rows: ${pruned.complete} complete sets, ${pruned.judged} judged.`) + if (pruned.rows) logger.info(`Pruned ${pruned.rows} par2 file rows: ${pruned.complete} complete sets, ${pruned.judged} judged, ${pruned.final} final.`) } 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).`) @@ -335,6 +335,12 @@ export class Par2Merger { 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]) } + // An unsettled member's candidates into a settled target: the settler won't come by again to make them claimable. + await trx.rawQuery( + `UPDATE probe_candidates pc SET ready = true, unreadable = NOT r.readable FROM releases r + WHERE r.id = ? AND r.settled_at IS NOT NULL AND pc.release_id = r.id AND pc.status = 'waiting' AND NOT pc.ready`, + [target.id] + ) // A set held by the target already, or by two of the others (a par2 and its repost), isn't held twice (every // round failed on the primary key from 20:28 UTC on 2026-10-12; LOG.md 2026-10-12 (30)). await trx.rawQuery( diff --git a/app/services/par2_pruner.ts b/app/services/par2_pruner.ts index 5e4423c..4d839ef 100644 --- a/app/services/par2_pruner.ts +++ b/app/services/par2_pruner.ts @@ -4,86 +4,98 @@ import db from '@adonisjs/lucid/services/db' const SETTLED_MS = 2 * 86400_000 /** Sets per statement. */ const BATCH = 20_000 -/** The complete pass's row in worker_cursors: the newest holder's created_at of the last set it looked at. */ +/** The sweep's row in worker_cursors: the last set_id it looked at (key); past the last set it starts over. */ const CURSOR = 'par2-prune' +/** The merger's row in worker_cursors: the end of its last whole window. */ +const MERGER = 'merger' +/** The merger's windows look back this far (Par2Merger's OVERLAP_MS). */ +const MERGER_OVERLAP_MS = 60_000 /** Verdicts whose releases needn't merge any more; locked isn't one (a password found takes it back). */ const JUNK = ['dead', 'fake', 'repeated'] +/** Probe candidates whose results a merge reads: a file's hash (lone), a set's file list (par2). */ +const MERGE_KINDS = ['lone', 'par2'] export interface PruneRound { complete: number judged: number + final: number rows: number } /** - * Drops par2_files rows the merge can't use any more (LOG.md 2026-10-16 (60)): the table only grew, 35M rows and - * 13.6 GB at 1.8M sets. A set's file list matches file hashes to releases and names an unreadable upload once; it's - * done when + * Drops par2_files rows the merge can't use any more (LOG.md 2026-10-16 (60), 2026-10-20 (87)). A set's file list + * matches file hashes to releases and names an unreadable upload once; it's done when * - complete: SETTLED_MS after its last holder's par2, every file it describes has its hash in a release holding the * set (all of it merged into one); - * - judged: every release holding it has a junk verdict (dead, fake, repeated). + * - judged: every release holding it has a junk verdict (dead, fake, repeated); + * - final: SETTLED_MS after its last holder's par2, every holder is final (release_final: a run of done ranges holds + * it with hours to spare, so no more of its upload will be read), the candidates of the ranges around it are probed, + * and the merger has seen every hash of its files. A file still missing then is one no probe will hash: most sets + * describe a readable upload's files, which aren't lone candidates (9 in 10 sets were kept for that, 19 GB). * par2_sets stays (the judge's unusable rule, the merge's holders). A set comes up again only with a new holder: its * par2 reposted, whose probe writes the rows again; without rows its merge finds no files and the count check holds. + * The sweep goes through the sets with rows by set_id and starts over at the end: a set kept now (a holder at the edge + * of a run, its ranges' probes not done yet) is looked at again on the next pass. */ export class Par2Pruner { /** Sets per batch (BATCH; smaller in the tests). */ constructor(private batch = BATCH) {} /** - * Batches until done or `budgetMs` is spent (the cursor is saved per batch, the next round goes on from it): run - * from the merger's loop, a first round over two days of sets held the merges back for hours (LOG.md 2026-10-17 (72)). + * Batches until a pass is done or `budgetMs` is spent (the cursor is saved per batch, the next round goes on from + * it): run from the merger's loop, a first round over two days of sets held the merges back for hours (LOG.md + * 2026-10-17 (72)). */ async round(budgetMs = Infinity): Promise { const deadline = Date.now() + budgetMs - const round: PruneRound = { complete: 0, judged: 0, rows: 0 } - // (created_at to the microsecond, set_id): one prober flush gives thousands of sets the same created_at, and a - // timestamp read back to the millisecond took the same batch again every round (LOG.md 2026-10-17 (73)). - const { rows: cursors } = await db.rawQuery(`SELECT at::text AS at, key FROM worker_cursors WHERE name = ?`, [CURSOR]) - let since: string = cursors[0]?.at ?? '1970-01-01 00:00:00+00' - let sinceKey: string = cursors[0]?.key ?? '' - const until = new Date(Date.now() - SETTLED_MS) + const round: PruneRound = { complete: 0, judged: 0, final: 0, rows: 0 } + const { rows: cursors } = await db.rawQuery(`SELECT key FROM worker_cursors WHERE name = ?`, [CURSOR]) + let after: string = cursors[0]?.key ?? '' + // Ranges done since the oldest merge-kind candidate still unprobed: their releases' hashes and file lists may be + // still to come. Hashes since the merger's last window: it hasn't matched them yet. + const { rows: marks } = await db.rawQuery( + `SELECT coalesce((SELECT min(created_at) FROM probe_candidates WHERE status IN ('waiting', 'taken') AND kind = ANY(?)), now()) AS probed, + coalesce((SELECT at FROM worker_cursors WHERE name = ?), to_timestamp(0)) - make_interval(secs => ?) AS merged, + now() - make_interval(secs => ?) AS settled`, + [MERGE_KINDS, MERGER, MERGER_OVERLAP_MS / 1000, SETTLED_MS / 1000] + ) + const { probed, merged, settled } = marks[0] for (;;) { const { rows: sets } = await db.rawQuery( - `SELECT set_id, max(created_at)::text AS at FROM par2_sets GROUP BY set_id - HAVING (max(created_at), set_id::text) > (?::timestamptz, ?::text) AND max(created_at) <= ? - ORDER BY max(created_at), set_id LIMIT ?`, - [since, sinceKey, until, this.batch] + `WITH batch AS MATERIALIZED (SELECT DISTINCT set_id FROM par2_files WHERE set_id > ? ORDER BY set_id LIMIT ?), + recent AS MATERIALIZED (SELECT group_id, posted_low, posted_high FROM ranges WHERE status = 'done' AND done_at >= ?), + unseen AS MATERIALIZED (SELECT DISTINCT hash16k, size FROM file_hashes WHERE created_at >= ?) + SELECT b.set_id, CASE + WHEN NOT EXISTS (SELECT 1 FROM par2_sets s WHERE s.set_id = b.set_id AND s.created_at > ?) + AND NOT EXISTS (SELECT 1 FROM par2_files f WHERE f.set_id = b.set_id AND NOT EXISTS ( + SELECT 1 FROM file_hashes h JOIN par2_sets s ON s.release_id = h.release_id AND s.set_id = f.set_id + WHERE h.hash16k = f.hash16k AND h.size = f.length)) THEN 'complete' + WHEN NOT EXISTS (SELECT 1 FROM par2_sets o LEFT JOIN release_verdicts v ON v.release_id = o.release_id AND v.rule = ANY(?) + WHERE o.set_id = b.set_id AND v.release_id IS NULL) THEN 'judged' + WHEN NOT EXISTS (SELECT 1 FROM par2_sets s WHERE s.set_id = b.set_id AND s.created_at > ?) + AND NOT EXISTS (SELECT 1 FROM par2_sets s JOIN releases r ON r.id = s.release_id WHERE s.set_id = b.set_id AND ( + release_final(r.group_id, r.first_article, r.last_article, r.posted_at, r.last_posted_at, r.complete) IS NOT TRUE + OR EXISTS (SELECT 1 FROM recent x WHERE x.group_id = r.group_id + AND x.posted_low <= coalesce(r.last_posted_at, r.posted_at) + interval '6 hours' AND x.posted_high >= r.posted_at - interval '6 hours'))) + AND NOT EXISTS (SELECT 1 FROM par2_files f JOIN unseen u ON u.hash16k = f.hash16k AND u.size = f.length WHERE f.set_id = b.set_id) + THEN 'final' END AS reason + FROM batch b ORDER BY b.set_id`, + [after, this.batch, probed, merged, settled, JUNK, settled] ) - if (sets.length === 0) break - const ids = sets.map((s: { set_id: string }) => s.set_id) - const { rows: done } = await db.rawQuery( - `SELECT c.set_id FROM unnest(?::char(32)[]) AS c(set_id) - WHERE EXISTS (SELECT 1 FROM par2_files f WHERE f.set_id = c.set_id) - AND NOT EXISTS ( - SELECT 1 FROM par2_files f WHERE f.set_id = c.set_id AND NOT EXISTS ( - SELECT 1 FROM file_hashes h JOIN par2_sets s ON s.release_id = h.release_id AND s.set_id = f.set_id - WHERE h.hash16k = f.hash16k AND h.size = f.length))`, - [ids] - ) - round.complete += done.length - round.rows += await this.drop(done.map((s: { set_id: string }) => s.set_id)) - since = sets.at(-1).at - sinceKey = sets.at(-1).set_id + const done = (reason: keyof PruneRound) => sets.filter((s: { reason: string }) => s.reason === reason).map((s: { set_id: string }) => s.set_id) + for (const reason of ['complete', 'judged', 'final'] as const) { + const ids = done(reason) + round[reason] += ids.length + round.rows += await this.drop(ids) + } + // Past the last set: the next batch starts the next pass. + after = sets.length < this.batch ? '' : sets.at(-1).set_id await db.rawQuery( - `INSERT INTO worker_cursors (name, at, key) VALUES (?, ?::timestamptz, ?) ON CONFLICT (name) DO UPDATE SET at = excluded.at, key = excluded.key`, - [CURSOR, since, sinceKey] + `INSERT INTO worker_cursors (name, at, key) VALUES (?, now(), ?) ON CONFLICT (name) DO UPDATE SET at = excluded.at, key = excluded.key`, + [CURSOR, after] ) - if (sets.length < this.batch || Date.now() > deadline) break + if (after === '' || Date.now() > deadline) break } - if (Date.now() > deadline) return round - // Judged: no cursor (verdicts come and go); only sets with rows left. - const { rows: judged } = await db.rawQuery( - `SELECT DISTINCT s.set_id FROM release_verdicts v JOIN par2_sets s ON s.release_id = v.release_id - WHERE v.rule = ANY(?) - AND EXISTS (SELECT 1 FROM par2_files f WHERE f.set_id = s.set_id) - AND NOT EXISTS ( - SELECT 1 FROM par2_sets o LEFT JOIN release_verdicts ov ON ov.release_id = o.release_id AND ov.rule = ANY(?) - WHERE o.set_id = s.set_id AND ov.release_id IS NULL)`, - [JUNK, JUNK] - ) - round.judged = judged.length - const judgedIds = judged.map((s: { set_id: string }) => s.set_id) - for (let i = 0; i < judgedIds.length; i += BATCH) round.rows += await this.drop(judgedIds.slice(i, i + BATCH)) return round } diff --git a/database/migrations/1770000005100_late_candidates_ready.ts b/database/migrations/1770000005100_late_candidates_ready.ts new file mode 100644 index 0000000..c40174b --- /dev/null +++ b/database/migrations/1770000005100_late_candidates_ready.ts @@ -0,0 +1,17 @@ +import { BaseSchema } from '@adonisjs/lucid/schema' + +/** + * Probe candidates added to a release after it settled (a later range of its upload, a merge into it) stayed not + * ready: the settler marks a release's candidates only when it settles it (LOG.md 2026-10-20 (87)). 93,642 waited + * since 2026-10-13 (17k lone, 31k par2, 45k archive). The batch store and the merge mark them now; these are made + * claimable once. + */ +export default class extends BaseSchema { + async up() { + this.schema.raw(` + UPDATE probe_candidates pc SET ready = true, unreadable = NOT r.readable FROM releases r + WHERE pc.status = 'waiting' AND NOT pc.ready AND r.id = pc.release_id AND r.settled_at IS NOT NULL`) + } + + async down() {} +} diff --git a/tests/functional/batch_store.spec.ts b/tests/functional/batch_store.spec.ts index 34e3691..0954d9c 100644 --- a/tests/functional/batch_store.spec.ts +++ b/tests/functional/batch_store.spec.ts @@ -120,6 +120,18 @@ test.group('BatchStore', (group) => { assert.equal(Number(chunks!.n), 4) }) + test('candidates added to a settled release (a later range of its upload) are claimable at once', async ({ assert }) => { + await storeIn(GROUP, 1, upload(10, 'Some.Show.S01E02.1080p.mkv', 2, 'Some.Show.S01E02.1080p [1/2]', '2026-10-05T10:00:00Z')) + // Settled before the next range came (the settler marks only what it settles). + await db.from('releases').update({ settled_at: new Date() }) + await db.from('probe_candidates').update({ ready: true }) + await storeIn(GROUP, 100_001, upload(100_010, 'Some.Show.S01E02.1080p.par2', 1, 'Some.Show.S01E02.1080p [2/2]', '2026-10-05T10:20:00Z')) + const candidates = await db.from('probe_candidates').orderBy('id').select('kind', 'ready') + assert.deepEqual(candidates.map((c) => c.kind), ['par2']) + assert.isTrue(candidates.every((c) => c.ready)) + assert.lengthOf(await releasesOf(), 1) + }) + test('the same upload in another group (multi-posted) is a release of its own', async ({ assert }) => { await storeIn(GROUP, 1, upload(10, 'Some.Show.S01E01.1080p.mkv', 2, 'Some.Show.S01E01.1080p [1/1]', '2026-10-05T10:00:00Z')) await storeIn('alt.binaries.other', 1, upload(10, 'Some.Show.S01E01.1080p.mkv', 2, 'Some.Show.S01E01.1080p [1/1]', '2026-10-05T10:01:00Z', 'alt.binaries.other')) diff --git a/tests/functional/par2_pruner.spec.ts b/tests/functional/par2_pruner.spec.ts index 8cafeae..ff80d0b 100644 --- a/tests/functional/par2_pruner.spec.ts +++ b/tests/functional/par2_pruner.spec.ts @@ -53,23 +53,53 @@ test.group('Par2Pruner', (group) => { const locked = await set('e', e, 2, null) await db.table('release_verdicts').insert({ release_id: e, rule: 'locked', rule_version: 1 }) - // Out of budget: the complete pass's batch, not the judged pass (the next round goes on). - assert.deepEqual(await new Par2Pruner().round(0), { complete: 1, judged: 0, rows: 2 }) - await db.from('worker_cursors').where('name', 'par2-prune').delete() - const round = await new Par2Pruner().round() - assert.deepEqual(round, { complete: 0, judged: 1, rows: 2 }) + // Final: its holder held by a run of done ranges with hours to spare, nothing hashed (a readable upload's files). + const f = await release() + const final = await set('f', f, 2, null) + const posted = new Date('2024-03-01T12:00:00Z') + const [group] = await db.table('groups').insert({ name: 'alt.binaries.pruned', priority: 1 }).returning('id') + const groupId = Number(group.id) + await db.from('releases').where('id', f).update({ group_id: groupId, posted_at: posted, last_posted_at: posted, first_article: 5000, last_article: 5100 }) + const [range] = await db + .table('ranges') + .insert({ group_id: groupId, start_id: 1, end_id: 10000, next_id: 10001, status: 'done', done_at: old, posted_low: new Date('2024-02-28T00:00:00Z'), posted_high: new Date('2024-03-03T00:00:00Z') }) + .returning('id') + await db.table('range_blocks').insert({ group_id: groupId, start_id: 1, end_id: 10000, posted_low: new Date('2024-02-28T00:00:00Z'), posted_high: new Date('2024-03-03T00:00:00Z'), at_group_start: false }) + // A lone candidate unprobed since before the range was done: its hash may still match. Kept. + await db.table('probe_candidates').insert({ id: 9001, release_id: b, kind: 'lone', message_id: '<9001@x>', file: 'x', status: 'waiting', created_at: new Date(old.getTime() - 3600_000) }) + + // Out of budget: one batch of two sets (a, b), the next round goes on. + assert.deepEqual(await new Par2Pruner(2).round(0), { complete: 1, judged: 0, final: 0, rows: 2 }) + assert.deepEqual(await new Par2Pruner(2).round(), { complete: 0, judged: 1, final: 0, rows: 2 }) assert.equal(await count(complete), 0) assert.equal(await count(apart), 2) assert.equal(await count(recent), 2) assert.equal(await count(dead), 0) assert.equal(await count(locked), 2) - assert.equal((await db.from('par2_sets').count('* as n'))[0].n, '5') + assert.equal(await count(final), 2) - // Again: the cursor is past them, nothing more. - assert.deepEqual(await new Par2Pruner().round(), { complete: 0, judged: 0, rows: 0 }) + // Probed: final. The next pass starts over from the first set. + await db.from('probe_candidates').where('id', 9001).update({ status: 'done' }) + assert.deepEqual(await new Par2Pruner(2).round(), { complete: 0, judged: 0, final: 1, rows: 2 }) + assert.equal(await count(final), 0) + assert.equal((await db.from('par2_sets').count('* as n'))[0].n, '6') + + // A hash of its files since the merger's last window: the merger hasn't matched it yet. Kept. + const g = await release() + const unseen = await set('g', g, 1, null) + await db.from('releases').where('id', g).update({ group_id: groupId, posted_at: posted, last_posted_at: posted, first_article: 6000, last_article: 6100 }) + await db.table('worker_cursors').insert({ name: 'merger', at: new Date(Date.now() - 3600_000) }) + await db.table('probe_candidates').insert({ id: 9002, release_id: await release(), kind: 'lone', message_id: '<9002@x>', file: 'x', status: 'done' }) + await db.table('file_hashes').insert({ candidate_id: 9002, release_id: 1, hash16k: 'g0'.padEnd(32, '0'), size: 1000 }) + assert.equal((await new Par2Pruner().round()).final, 0) + // Past the hash by more than the merger's overlap. + await db.from('worker_cursors').where('name', 'merger').update({ at: new Date(Date.now() + 120_000) }) + assert.equal((await new Par2Pruner().round()).final, 1) + assert.equal(await count(unseen), 0) + assert.isNotNull(range) }) - test('more sets of one created_at than a batch holds: each batch goes on past the last one', async ({ assert }) => { + test('sets of one created_at, a batch each: every one looked at', async ({ assert }) => { // One prober flush gives thousands of sets the same created_at (to the microsecond). const flush = new Date('2026-10-15T14:38:39.883Z') let candidate = 7000 @@ -88,6 +118,7 @@ test.group('Par2Pruner', (group) => { const round = await new Par2Pruner(1).round() assert.equal(round.complete, 3) assert.equal((await db.from('par2_files').whereIn('set_id', ['p'.repeat(32), 'q'.repeat(32), 'r'.repeat(32)]).count('* as n'))[0].n, '0') - assert.equal((await db.from('worker_cursors').where('name', 'par2-prune').firstOrFail()).key, 'r'.repeat(32)) + // The pass ended: the next one starts from the first set. + assert.equal((await db.from('worker_cursors').where('name', 'par2-prune').firstOrFail()).key, '') }) })