diff --git a/REBUILD.md b/REBUILD.md index 4510ac3..b40573e 100644 --- a/REBUILD.md +++ b/REBUILD.md @@ -238,6 +238,10 @@ 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. +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)). **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/par2_merger.ts b/app/services/par2_merger.ts index 99ad50d..5706433 100644 --- a/app/services/par2_merger.ts +++ b/app/services/par2_merger.ts @@ -15,17 +15,28 @@ const MAX_HOLDERS = 4 /** Members with more files than this many times the set's, plus SLACK_FILES (its par2 volumes), are more than one upload. */ const MAX_FILES_PER_SET_FILE = 3 const SLACK_FILES = 10 +/** + * A par2 file held by more file rows than this (the same first 16 KiB and size) is a common file, not one upload's: + * it matches nothing. One such file was in 190,380 uploads, and a set naming it loaded every one of them; the merger + * ran out of heap every few minutes (LOG.md 2026-10-20 (85)). The judge calls 20 releases repeated. + */ +const COMMON_FILE = 20 /** Between rounds. */ const ROUND_MS = 60_000 -/** A round looks back this far past the previous one's start (rows committed while it ran). */ +/** A window looks back this far past the previous one's end (rows committed after it was read). */ const OVERLAP_MS = 60_000 +/** + * A round goes through what's new in windows of this much (created_at), its place kept after each: a round behind by + * two days of the probe backlog (3.5M sets) never ended, and every restart started it over (LOG.md 2026-10-20 (85)). + */ +const WINDOW_MS = 60 * 60_000 /** The sidecar fold also runs this often inside a long par2 pass: its queue grows by the settler's pace. */ const FOLD_EVERY_MS = 5 * 60_000 /** Between par2_files prunes (complete and judged sets' file lists). */ const PRUNE_EVERY_MS = 60 * 60_000 /** A prune's share of the merger's time: the rest goes on in the next one. */ const PRUNE_BUDGET_MS = 2 * 60_000 -/** The merger's row in worker_cursors: the start of its last whole round, kept over a restart. */ +/** The merger's row in worker_cursors: the end of its last whole window, kept over a restart. */ const CURSOR = 'merger' export interface MergeRound { @@ -46,6 +57,12 @@ export interface MergeRound { */ const INNER_STEM = `substring(i.result->>'yencName' from '^([A-Za-z0-9]{16,64})\\.')` +/** The releases holding a par2 file f (its first 16 KiB and size), none when it's a common file (COMMON_FILE). */ +const HOLDERS = `SELECT c.release_id FROM ( + SELECT l.release_id, count(*) OVER () AS holders + FROM (SELECT h.release_id FROM file_hashes h WHERE h.hash16k = f.hash16k AND h.size = f.length LIMIT ${COMMON_FILE + 1}) l + ) c WHERE c.holders <= ${COMMON_FILE}` + interface Member { id: number release_key: string @@ -111,14 +128,18 @@ export class Par2Merger { } } - /** The sets with something new since the last round: a par2 found, or a hash of one of their files. */ + /** + * The sets and inner stems with something new since the last round (a par2 found, a hash of one of their files, a + * lone inspection), a window at a time. + */ async round(): Promise { - const { rows: now } = await db.rawQuery(`SELECT now() AS at`) + // A Date keeps milliseconds, now() microseconds: rounded up, a row made at now() is before the end. + const { rows: now } = await db.rawQuery(`SELECT now() + interval '1 millisecond' AS at`) + const end = new Date(now[0].at) if (!this.since) { const saved = await db.from('worker_cursors').where('name', CURSOR).first() this.since = saved ? new Date(saved.at) : new Date(0) } - const since = new Date(this.since.getTime() - OVERLAP_MS) // First: sidecars (.nfo, .sfv, .nzb) apart from their upload join it. A queue, not since: it needn't wait for the // par2 pass (hours when that starts over). const folded = await this.folder.round() @@ -132,52 +153,96 @@ export class Par2Merger { foldedAt = Date.now() if (more.folded) logger.info(`Folded ${more.folded} sidecars into their uploads (${more.sidecars} sidecar-only of ${more.looked} small releases).`) } - 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, stems: 0, stemMerges: 0, stemReleases: 0, folded } - for (const { set_id: setId } of rows) { + const round: MergeRound = { sets: 0, merges: 0, releases: 0, stems: 0, stemMerges: 0, stemReleases: 0, folded } + while (this.since < end) { if (this.stopped) return round - await foldAgain() - // One set that fails doesn't hold up the others (nor the next round, which would start from the same point). - const merged = await this.mergeSet(setId.trim()).catch((error) => { - logger.error(`Par2 merge of set ${setId.trim()} failed: ${error.message}`) - return 0 - }) - if (merged > 0) { - round.merges++ - round.releases += merged + 1 + const from = await this.firstNew(new Date(this.since.getTime() - OVERLAP_MS), end) + const until = from ? new Date(Math.min(from.getTime() + WINDOW_MS, end.getTime())) : end + if (from) { + const { touched, sets } = await this.setsToMerge(from, until) + round.sets += touched + for (const setId of sets) { + if (this.stopped) return round + await foldAgain() + // One set that fails doesn't hold up the others (nor the next round, which would start from the same point). + const merged = await this.mergeSet(setId.trim()).catch((error) => { + logger.error(`Par2 merge of set ${setId.trim()} failed: ${error.message}`) + return 0 + }) + if (merged > 0) { + round.merges++ + round.releases += merged + 1 + } + } + const { rows: stems } = await db.rawQuery( + `SELECT DISTINCT ${INNER_STEM} AS stem FROM release_inspections i + WHERE i.kind = 'lone' AND i.created_at >= ? AND i.created_at <= ? AND ${INNER_STEM} IS NOT NULL`, + [from, until] + ) + for (const { stem } of stems) { + if (this.stopped) return round + if (!looksRandom(stem)) continue + round.stems++ + const merged = await this.mergeStem(stem).catch((error) => { + logger.error(`Inner-name merge of ${stem} failed: ${error.message}`) + return 0 + }) + if (merged > 0) { + round.stemMerges++ + round.stemReleases += merged + 1 + } + } } + // Only after a whole window: a restart halfway goes on from the window before. + this.since = until + await db.rawQuery( + `INSERT INTO worker_cursors (name, at) VALUES (?, ?) ON CONFLICT (name) DO UPDATE SET at = excluded.at`, + [CURSOR, this.since] + ) } - const { rows: stems } = await db.rawQuery( - `SELECT DISTINCT ${INNER_STEM} AS stem FROM release_inspections i WHERE i.kind = 'lone' AND i.created_at >= ? AND ${INNER_STEM} IS NOT NULL`, - [since] - ) - for (const { stem } of stems) { - if (this.stopped) return round - if (!looksRandom(stem)) continue - round.stems++ - const merged = await this.mergeStem(stem).catch((error) => { - logger.error(`Inner-name merge of ${stem} failed: ${error.message}`) - return 0 - }) - if (merged > 0) { - round.stemMerges++ - round.stemReleases += merged + 1 - } - } - // Only after a whole round: a restart halfway goes on from the round before. - this.since = now[0].at - await db.rawQuery( - `INSERT INTO worker_cursors (name, at) VALUES (?, ?) ON CONFLICT (name) DO UPDATE SET at = excluded.at`, - [CURSOR, this.since] - ) return round } + /** When the first row a window reads was made, from `from` up to `end`; null when none (a gap, the nightly pause). */ + private async firstNew(from: Date, end: Date): Promise { + const { rows } = await db.rawQuery( + `SELECT least( + (SELECT min(created_at) FROM par2_sets WHERE created_at >= ?), + (SELECT min(created_at) FROM file_hashes WHERE created_at >= ?), + (SELECT min(created_at) FROM release_inspections WHERE created_at >= ?)) AS at`, + [from, from, from] + ) + const at = rows[0].at ? new Date(rows[0].at) : null + return at && at <= end ? at : null + } + + /** + * The sets a window touched (a par2 found, a hash of one of their files), and those among them a merge could + * change: held or matched by two releases or more. Most have every file in one release already (96% of a backlog + * hour's 147k): mergeSet needn't look at them. + */ + private async setsToMerge(from: Date, until: Date): Promise<{ touched: number; sets: string[] }> { + const { rows } = await db.rawQuery( + `WITH touched AS MATERIALIZED ( + SELECT set_id FROM par2_sets WHERE created_at >= ? AND 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 >= ? AND h.created_at <= ? + ) + SELECT (SELECT count(*) FROM touched) AS touched, array( + SELECT t.set_id FROM touched t + WHERE EXISTS ( + SELECT 1 FROM ( + SELECT s.release_id FROM par2_sets s WHERE s.set_id = t.set_id + UNION ALL + SELECT m.release_id FROM par2_files f CROSS JOIN LATERAL (${HOLDERS}) m WHERE f.set_id = t.set_id + ) u HAVING count(DISTINCT u.release_id) >= 2) + ) AS sets`, + [from, until, from, until] + ) + return { touched: Number(rows[0].touched), sets: rows[0].sets } + } + /** Merges the releases whose lone files' inner names share a random stem; how many went into another. */ async mergeStem(stem: string): Promise { return db.transaction(async (trx) => { @@ -222,7 +287,7 @@ export class Par2Merger { const near = (r: Member) => r.group_id === anchor.group_id && (at(r) === null || at(anchor) === null || Math.abs(at(r)! - at(anchor)!) <= MAX_SPREAD_MS) const { rows: matches } = await trx.rawQuery( `SELECT f.file_id, r.id, r.release_key, r.group_id, r.files, r.posted_at, r.readable - FROM par2_files f JOIN file_hashes h ON h.hash16k = f.hash16k AND h.size = f.length JOIN releases r ON r.id = h.release_id + FROM par2_files f CROSS JOIN LATERAL (${HOLDERS}) h JOIN releases r ON r.id = h.release_id WHERE f.set_id = ?`, [setId] ) diff --git a/tests/functional/par2_merger.spec.ts b/tests/functional/par2_merger.spec.ts index 22b10fa..1019d22 100644 --- a/tests/functional/par2_merger.spec.ts +++ b/tests/functional/par2_merger.spec.ts @@ -78,6 +78,44 @@ test.group('Par2Merger', (group) => { assert.equal((await new Par2Merger().round()).sets, 1) }) + test('a round behind by hours goes through them a window at a time, past the gaps', async ({ assert }) => { + const ago = (hours: number) => new Date(Date.now() - hours * 3600_000) + await db.table('par2_sets').multiInsert([ + { set_id: 'a1'.repeat(16), release_id: 1, files: 1, created_at: ago(30) }, + { set_id: 'a2'.repeat(16), release_id: 2, files: 1, created_at: ago(2.5) }, + { set_id: 'a3'.repeat(16), release_id: 3, files: 1, created_at: ago(0.5) }, + ]) + await db.table('worker_cursors').insert({ name: 'merger', at: ago(48) }) + assert.equal((await new Par2Merger().round()).sets, 3) + const saved = await db.from('worker_cursors').where('name', 'merger').firstOrFail() + assert.isAbove(new Date(saved.at).getTime(), ago(0.5).getTime()) + }) + + test('a par2 file held by more than 20 files is a common one: it merges nothing', async ({ assert }) => { + const at = new Date('2022-07-08T03:46:00Z') + const ids: number[] = [] + for (const key of ['dd000001', 'dd000002']) { + 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: 1, files: 1, posted_at: at, readable: false }) + .returning('id') + ids.push(Number(row.id)) + } + // The par2 is held by the first; the second holds the file, and so do 20 releases elsewhere and years apart. + const setId = hash('1') + await db.table('par2_sets').insert({ set_id: setId, release_id: ids[0], files: 1 }) + await db.table('par2_files').insert({ set_id: setId, file_id: hash('0'), hash16k: hash('9'), length: 30214, name: 'sample.nfo' }) + await db.table('file_hashes').insert({ candidate_id: 4000, release_id: ids[1], hash16k: hash('9'), size: 30214 }) + for (let i = 1; i <= 20; i++) await db.table('file_hashes').insert({ candidate_id: 4000 + i, release_id: 900000 + i, hash16k: hash('9'), size: 30214 }) + const round = await new Par2Merger().round() + assert.deepEqual([round.sets, round.merges], [1, 0]) + assert.equal(await new Par2Merger().mergeSet(setId), 0) + assert.equal((await db.from('releases').whereIn('id', ids).select('id')).length, 2) + // One holder fewer: the file is one upload's again. + await db.from('file_hashes').where('candidate_id', 4020).delete() + assert.equal(await new Par2Merger().mergeSet(setId), 1) + }) + test('a set held by two of the releases merged is held once by the target', async ({ assert }) => { const at = new Date('2022-07-08T03:46:00Z') const ids: number[] = []