diff --git a/app/services/par2_merger.ts b/app/services/par2_merger.ts index f531399..8ab9f94 100644 --- a/app/services/par2_merger.ts +++ b/app/services/par2_merger.ts @@ -22,8 +22,18 @@ export interface MergeRound { sets: number merges: number releases: number + /** The inner-name pass: stems checked, merges, releases in them. */ + stems: number + stemMerges: number + stemReleases: number } +/** + * A lone inspection's inner stem: its yEnc name up to the first dot, when 16–64 letters and digits + * ("bX147zx8I0oEgj7nzFCiorCT0ps2jEcb.7z.001"). The same expression as release_inspections_inner_stem_idx. + */ +const INNER_STEM = `substring(i.result->>'yencName' from '^([A-Za-z0-9]{16,64})\\.')` + interface Member { id: number release_key: string @@ -31,6 +41,8 @@ interface Member { files: number posted_at: Date | null readable: boolean + poster?: string | null + file_total?: number | null } /** @@ -45,6 +57,15 @@ interface Member { * 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. + * + * A second pass merges by inner name: the yEnc name in a lone file's first + * article keeps the upload's random stem when the subject names each file + * apart ("[1/7] - "471044d8…"" holds "bX147zx8I0oEgj7nzFCiorCT0ps2jEcb.7z.001"; + * hep, hephaestus, gyges: every stem under one counter total; LOG.md + * 2026-10-12 (32)). Releases whose lone inspections share such a stem, in one + * group, by one poster, under the same counter total and within MAX_SPREAD_MS, + * are one upload. Only random stems: a readable one is grouped from the + * headers already, and a repost would share it. */ export class Par2Merger { private stopped = false @@ -60,6 +81,7 @@ export class Par2Merger { 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).`) + if (round.stemMerges > 0) logger.info(`Merged ${round.stemReleases} releases into ${round.stemMerges} by inner name (${round.stems} stems checked).`) } catch (error: any) { logger.error(`Par2 merge round failed: ${error.message}`) } @@ -77,7 +99,7 @@ export class Par2Merger { 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 } + const round: MergeRound = { sets: rows.length, merges: 0, releases: 0, stems: 0, stemMerges: 0, stemReleases: 0 } for (const { set_id: setId } of rows) { if (this.stopped) return round // One set that fails doesn't hold up the others (nor the next round, which would start from the same point). @@ -90,10 +112,57 @@ export class Par2Merger { 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 ${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 + } + } this.since = now[0].at return round } + /** 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) => { + const { rows } = await trx.rawQuery( + `SELECT DISTINCT r.id, r.release_key, r.group_id, r.files, r.posted_at, r.readable, r.poster, r.file_total + FROM release_inspections i JOIN releases r ON r.id = i.release_id + WHERE i.kind = 'lone' AND ${INNER_STEM} = ?`, + [stem] + ) + const sorted: Member[] = rows.sort((a: Member, b: Member) => (a.posted_at ? new Date(a.posted_at).getTime() : 0) - (b.posted_at ? new Date(b.posted_at).getTime() : 0)) + const anchor = sorted[0] + if (!anchor) return 0 + const at = (r: Member) => (r.posted_at ? new Date(r.posted_at).getTime() : null) + const total = (r: Member) => (r.file_total ? Number(r.file_total) : null) + const near = (r: Member) => + r.group_id === anchor.group_id && + (r.poster ?? '') === (anchor.poster ?? '') && + (total(r) === null || total(anchor) === null || total(r) === total(anchor)) && + (at(r) === null || at(anchor) === null || Math.abs(at(r)! - at(anchor)!) <= MAX_SPREAD_MS) + const found = new Map() + for (const r of sorted.filter(near)) found.set(String(r.id), r) + if (found.size < 2) return 0 + // No more files than the counter says, with room for par2 volumes posted under it. + const counted = Math.max(0, ...[...found.values()].map((r) => total(r) ?? 0)) + const files = [...found.values()].reduce((n, m) => n + Number(m.files), 0) + if (counted > 0 && files > counted + SLACK_FILES) return 0 + return this.merge(trx, found, stem, 'inner') + }) + } + /** Merges the releases of one set; how many went into another. */ async mergeSet(setId: string): Promise { return db.transaction(async (trx) => { @@ -127,21 +196,31 @@ export class Par2Merger { for (const r of nearHolders) found.set(String(r.id), r) for (const releases of byFile.values()) if (releases.size === 1) for (const [id, r] of releases) found.set(id, r) if (found.size < 2) return 0 + const { rows: described } = await trx.rawQuery(`SELECT count(*) AS n FROM par2_files WHERE set_id = ?`, [setId]) + const setFiles = Math.max(Number(described[0].n), 1) + if ([...found.values()].reduce((n, m) => n + Number(m.files), 0) > MAX_FILES_PER_SET_FILE * setFiles + SLACK_FILES) return 0 + return this.merge(trx, found, setId, 'par2') + }) + } + + /** + * Merges the found releases into the one with the most files (named by the par2 set's files when none + * is readable); how many went into it. Each merge is kept in release_merges, with its set or stem. + */ + private async merge(trx: any, found: Map, key: string, via: 'par2' | 'inner'): Promise { + { // Locked in id order (the batch store and other rounds take the same rows). const { rows: locked } = await trx.rawQuery(`SELECT id FROM releases WHERE id = ANY(?) ORDER BY id FOR UPDATE`, [[...found.keys()]]) const members: Member[] = locked.map((r: { id: string }) => found.get(String(r.id))!) if (members.length < 2) return 0 - const { rows: described } = await trx.rawQuery(`SELECT count(*) AS n FROM par2_files WHERE set_id = ?`, [setId]) - const setFiles = Math.max(Number(described[0].n), 1) - if (members.reduce((n, m) => n + Number(m.files), 0) > MAX_FILES_PER_SET_FILE * setFiles + SLACK_FILES) 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 + const name = via === 'par2' && members.every((m) => !m.readable) ? await this.setName(trx, key) : 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] + `INSERT INTO release_merges (from_id, from_key, into_id, set_id, files, via) + SELECT id, release_key, ?, ?, files, ? FROM releases WHERE id = ANY(?)`, + [target.id, key, via, 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]) @@ -180,7 +259,7 @@ export class Par2Merger { ) 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. */ diff --git a/database/migrations/1770000002600_inner_stem_merge.ts b/database/migrations/1770000002600_inner_stem_merge.ts new file mode 100644 index 0000000..b7014d2 --- /dev/null +++ b/database/migrations/1770000002600_inner_stem_merge.ts @@ -0,0 +1,30 @@ +import { BaseSchema } from '@adonisjs/lucid/schema' + +/** + * The merge by inner name (app/services/par2_merger.ts, InnerStemMerger): + * obfuscated uploads post each file under a subject name of its own, but the + * yEnc name inside its first article shares a random stem with the rest of + * the upload ("[1/7] - "471044d8…"" holds "bX147zx8I0oEgj7nzFCiorCT0ps2jEcb.7z.001"; + * hep, hephaestus, gyges: 31,000 one-file releases; LOG.md 2026-10-12 (32)). + * - An index over a lone inspection's stem: the yEnc name up to its first dot, when 16–64 letters and digits. + * - release_merges: which kind of merge (par2 or inner), and room for a stem in set_id. + */ +export default class extends BaseSchema { + static disableTransactions = true + + async up() { + this.schema.raw( + `CREATE INDEX CONCURRENTLY IF NOT EXISTS release_inspections_inner_stem_idx ON release_inspections + ((substring(result->>'yencName' from '^([A-Za-z0-9]{16,64})\\.'))) WHERE kind = 'lone'` + ) + this.schema.raw(`CREATE INDEX CONCURRENTLY IF NOT EXISTS release_inspections_created_idx ON release_inspections (created_at)`) + this.schema.raw(`ALTER TABLE release_merges ALTER COLUMN set_id TYPE varchar(64)`) + this.schema.raw(`ALTER TABLE release_merges ADD COLUMN IF NOT EXISTS via varchar(8) NOT NULL DEFAULT 'par2'`) + } + + async down() { + this.schema.raw(`DROP INDEX CONCURRENTLY IF EXISTS release_inspections_inner_stem_idx`) + this.schema.raw(`DROP INDEX CONCURRENTLY IF EXISTS release_inspections_created_idx`) + this.schema.raw(`ALTER TABLE release_merges DROP COLUMN IF EXISTS via`) + } +} diff --git a/tests/functional/par2_merger.spec.ts b/tests/functional/par2_merger.spec.ts index 20ffa14..168dd92 100644 --- a/tests/functional/par2_merger.spec.ts +++ b/tests/functional/par2_merger.spec.ts @@ -47,7 +47,7 @@ test.group('Par2Merger', (group) => { 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 }) + assert.deepEqual(round, { sets: 1, merges: 1, releases: 3, stems: 0, stemMerges: 0, stemReleases: 0 }) const target = await db.from('releases').where('id', b).firstOrFail() assert.equal(target.files, 4) assert.equal(target.segments, 40) @@ -104,6 +104,62 @@ test.group('Par2Merger', (group) => { assert.equal((await db.from('releases').whereIn('id', ids).select('id')).length, 6) }) + test('merges the one-file releases of an obfuscated upload by the random stem of their inner names', async ({ assert }) => { + const at = new Date('2022-07-07T19:34:00Z') + let n = 0 + const release = async (poster: string, posted: Date, total: number, inner: string | null, groupId = 1) => { + n++ + const key = `dd${String(n).padStart(6, '0')}` + const [row] = await db + .table('releases') + .insert({ release_key: key.padEnd(40, '0'), identity_key: key.padEnd(40, '0'), name: `hex${n}`, source: 'stem', groups: '[]', group_id: groupId, files: 1, posted_at: posted, readable: false, file_total: total, poster }) + .returning('id') + if (inner) { + await db.table('release_inspections').insert({ release_id: row.id, candidate_id: 5000 + n, kind: 'lone', result: JSON.stringify({ yencName: inner, type: '', size: 1 }) }) + } + return Number(row.id) + } + const hep = 'hep ' + const stem = 'nabNGjSlKhOYJCslnEw1MzJIgPIjffyT' + const a = await release(hep, at, 15, `${stem}.7z.001`) + const b = await release(hep, new Date(at.getTime() + 60_000), 15, `${stem}.7z.002`) + const c = await release(hep, new Date(at.getTime() + 120_000), 15, `${stem}.vol07+08.par2`) + // Another poster, a week later, another total, another group: apart. + const other = await release('someone ', at, 15, `${stem}.7z.003`) + const later = await release(hep, new Date(at.getTime() + 7 * 86400_000), 15, `${stem}.7z.004`) + const total = await release(hep, at, 16, `${stem}.7z.005`) + const group = await release(hep, at, 15, `${stem}.7z.006`, 2) + // A readable stem isn't merged on: a repost would share it. + const r1 = await release(hep, at, 15, 'SomeReadableMovieName.7z.001') + const r2 = await release(hep, at, 15, 'SomeReadableMovieName.7z.002') + + const round = await new Par2Merger().round() + assert.equal(round.stemMerges, 1) + assert.equal(round.stemReleases, 3) + const left = (await db.from('releases').whereIn('id', [a, b, c]).select('id', 'files')).map((r) => Number(r.files)) + assert.deepEqual(left, [3]) + assert.equal((await db.from('releases').whereIn('id', [other, later, total, group, r1, r2]).select('id')).length, 6) + const merges = await db.from('release_merges').where('via', 'inner').select('set_id') + assert.deepEqual([...new Set(merges.map((m) => m.set_id))], [stem]) + }) + + test('an inner-name merge with more files than the counter is more than one upload', async ({ assert }) => { + const at = new Date('2022-07-07T19:34:00Z') + const stem = 'R8aUMXqWQfOaei4u74fch9v0skxUY7ah' + const ids: number[] = [] + for (let i = 1; i <= 3; i++) { + const key = `ee00000${i}` + 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: 10, posted_at: at, readable: false, file_total: 5, poster: 'p' }) + .returning('id') + ids.push(Number(row.id)) + await db.table('release_inspections').insert({ release_id: row.id, candidate_id: 6000 + i, kind: 'lone', result: JSON.stringify({ yencName: `${stem}.7z.00${i}` }) }) + } + assert.equal(await new Par2Merger().mergeStem(stem), 0) + assert.equal((await db.from('releases').whereIn('id', ids).select('id')).length, 3) + }) + test('a set file that matches several releases nearby is ambiguous: not merged', async ({ assert }) => { const at = new Date('2022-07-08T03:46:00Z') const ids: number[] = []