diff --git a/REBUILD.md b/REBUILD.md index 0a68e0d..a376515 100644 --- a/REBUILD.md +++ b/REBUILD.md @@ -272,6 +272,15 @@ spot, a day after settling). Verdicts (`release_verdicts`, with the rule set's version) hide releases from search (`judged` shows them); a verdict whose rule no longer matches is taken back. Nothing is dropped yet. +**Sidecars** (the merger's third pass). An upload's .nfo, .sfv or .nzb +that came apart from it (another name, another counter: "[00/80] … .nzb" +beside "[01/94] … .part01.rar") was a release of its own. The settler +queues small settled releases (≤ 3 files, < 1 MB) in `fold_candidates`; +one whose files are all sidecars joins the poster's bigger release in the +same group posted within 10 minutes: the one sharing a word of 4+ letters +with its subject, else the one under the same counter total, else none +(`release_merges.via` 'sidecar'). + **Spots** (after voyager 1's D29; `lib/spotnet` ported as is). Two processes, one NNTP connection each: diff --git a/app/services/par2_merger.ts b/app/services/par2_merger.ts index 8ab9f94..2b9cde0 100644 --- a/app/services/par2_merger.ts +++ b/app/services/par2_merger.ts @@ -2,6 +2,7 @@ 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' +import { SidecarFolder } from '#services/sidecar_folder' /** 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 @@ -26,6 +27,8 @@ export interface MergeRound { stems: number stemMerges: number stemReleases: number + /** The sidecar fold: candidates looked at, sidecar-only ones among them, folded into their upload. */ + folded?: { looked: number; sidecars: number; folded: number } } /** @@ -69,6 +72,7 @@ interface Member { */ export class Par2Merger { private stopped = false + private folder = new SidecarFolder((trx, found, key, via) => this.merge(trx, found, key, via)) private since = new Date(0) stop() { @@ -82,6 +86,7 @@ export class Par2Merger { 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).`) + if (round.folded?.folded) logger.info(`Folded ${round.folded.folded} sidecars into their uploads (${round.folded.sidecars} sidecar-only of ${round.folded.looked} small releases).`) } catch (error: any) { logger.error(`Par2 merge round failed: ${error.message}`) } @@ -129,6 +134,8 @@ export class Par2Merger { round.stemReleases += merged + 1 } } + // Third: sidecars (.nfo, .sfv, .nzb) apart from their upload join it. + if (!this.stopped) round.folded = await this.folder.round() this.since = now[0].at return round } @@ -207,7 +214,7 @@ export class Par2Merger { * 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 { + private async merge(trx: any, found: Map, key: string, via: 'par2' | 'inner' | 'sidecar'): 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()]]) diff --git a/app/services/settler.ts b/app/services/settler.ts index 7d83548..ab148ba 100644 --- a/app/services/settler.ts +++ b/app/services/settler.ts @@ -27,8 +27,11 @@ class Settler { UPDATE releases SET settled_at = now(), complete = CASE WHEN file_total IS NULL THEN NULL 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), - marked AS (UPDATE probe_candidates pc SET ready = true, unreadable = NOT s.readable FROM settled s WHERE pc.release_id = s.id) + RETURNING id, readable, files, bytes, poster), + marked AS (UPDATE probe_candidates pc SET ready = true, unreadable = NOT s.readable FROM settled s WHERE pc.release_id = s.id), + -- Small ones may be an upload's .nfo or .nzb that came apart from it: the merger's sidecar fold looks at them. + queued AS (INSERT INTO fold_candidates (release_id) SELECT id FROM settled WHERE files <= 3 AND bytes < 1000000 AND poster IS NOT NULL + ON CONFLICT DO NOTHING) SELECT count(*)::int AS n FROM settled`, [QUIET_HISTORY_MINUTES, QUIET_HOURS, QUIET_HOURS, BATCH] ) diff --git a/app/services/sidecar_folder.ts b/app/services/sidecar_folder.ts new file mode 100644 index 0000000..c95951a --- /dev/null +++ b/app/services/sidecar_folder.ts @@ -0,0 +1,91 @@ +import db from '@adonisjs/lucid/services/db' +import { decodeChunk, type ChunkFile } from '#lib/chunks/index' + +/** Candidates looked at per round. */ +const BATCH = 2000 +/** An upload's sidecar is posted with it: its release is looked for this far from it. */ +const WINDOW_MINUTES = 10 +/** The files an upload carries beside its content: an index, a checksum list, a scene release's own files. */ +const SIDECAR = /\.(nfo|sfv|nzb|srr|srs|txt|m3u|m3u8|diz|md5)$/i + +/** The words of a name worth comparing: 4+ letters or digits. */ +function words(text: string): Set { + return new Set(text.toLowerCase().split(/[^a-z0-9]+/).filter((w) => w.length >= 4)) +} + +export interface Sibling { + id: number + name: string + files: number + file_total: number | null +} + +/** + * Which of the nearby releases a sidecar belongs to: the one sharing a word of 4+ letters with the sidecar's subject + * ("Murdoch Mysteries S8D1 [00/80] - "Murdoch Mysteries S8D1.nzb"" beside "Murdoch Mysteries S8D1"), else the one + * under the same counter total ("(ALADIN) [01/34] - "bud.txt"" beside the poster's 34-file upload); none when that + * isn't one release. + */ +export function pickSibling(files: ChunkFile[], fileTotal: number | null, siblings: Sibling[]): Sibling | null { + const said = words(files.map((f) => f.subject).join(' ')) + const named = siblings.filter((s) => [...words(s.name)].some((w) => said.has(w))) + if (named.length === 1) return named[0] + if (named.length > 1) return null + const counted = fileTotal ? siblings.filter((s) => s.file_total === fileTotal) : [] + return counted.length === 1 ? counted[0] : null +} + +export interface FoldRound { + looked: number + sidecars: number + folded: number +} + +/** + * The sidecar fold (REBUILD.md, junk): an .nfo, .sfv or .nzb that came apart from its upload (another name, another + * counter: "[00/80] … .nzb" beside "[01/94] … .part01.rar") was a release of its own. The settler queues small + * settled releases (fold_candidates); a candidate whose files are all sidecars joins the poster's bigger release in + * the same group posted within WINDOW_MINUTES, the one pickSibling chooses, through the merger's merge (via + * 'sidecar'). The rest leave the queue untouched. + */ +export class SidecarFolder { + constructor(private merge: (trx: any, found: Map, key: string, via: 'sidecar') => Promise) {} + + async round(): Promise { + const round: FoldRound = { looked: 0, sidecars: 0, folded: 0 } + const { rows: queued } = await db.rawQuery( + `SELECT r.id, r.release_key, r.group_id, r.poster, r.posted_at, r.files, r.file_total, r.readable + FROM fold_candidates f JOIN releases r ON r.id = f.release_id ORDER BY f.created_at LIMIT ?`, + [BATCH] + ) + const { rows: stale } = await db.rawQuery( + `SELECT f.release_id FROM fold_candidates f LEFT JOIN releases r ON r.id = f.release_id WHERE r.id IS NULL LIMIT ?`, + [BATCH] + ) + const done: number[] = stale.map((r: { release_id: string }) => Number(r.release_id)) + for (const c of queued) { + round.looked++ + done.push(Number(c.id)) + const { rows: chunks } = await db.rawQuery(`SELECT payload FROM release_chunks WHERE release_id = ?`, [c.id]) + const files: ChunkFile[] = chunks.flatMap((x: { payload: Buffer }) => decodeChunk(x.payload)) + if (files.length === 0 || !files.every((f) => SIDECAR.test(f.name))) continue + round.sidecars++ + if (!c.posted_at) continue + const { rows: siblings } = await db.rawQuery( + `SELECT r.id, r.release_key, r.group_id, r.files, r.file_total, r.posted_at, r.readable, r.name FROM releases r + WHERE r.posted_at BETWEEN ?::timestamptz - make_interval(mins => ?) AND ?::timestamptz + make_interval(mins => ?) + AND r.group_id = ? AND r.poster = ? AND r.id <> ? AND r.files > ?`, + [c.posted_at, WINDOW_MINUTES, c.posted_at, WINDOW_MINUTES, c.group_id, c.poster, c.id, c.files] + ) + const sibling = pickSibling(files, c.file_total ? Number(c.file_total) : null, siblings.map((s: any) => ({ ...s, id: Number(s.id), files: Number(s.files), file_total: s.file_total === null ? null : Number(s.file_total) }))) + if (!sibling) continue + const full = siblings.find((s: any) => Number(s.id) === sibling.id) + const merged = await db.transaction((trx) => + this.merge(trx, new Map([[String(c.id), c], [String(full.id), full]]), `sidecar:${full.id}`, 'sidecar') + ) + if (merged > 0) round.folded++ + } + if (done.length > 0) await db.from('fold_candidates').whereIn('release_id', done).delete() + return round + } +} diff --git a/database/migrations/1770000003400_fold_candidates.ts b/database/migrations/1770000003400_fold_candidates.ts new file mode 100644 index 0000000..f242746 --- /dev/null +++ b/database/migrations/1770000003400_fold_candidates.ts @@ -0,0 +1,20 @@ +import { BaseSchema } from '@adonisjs/lucid/schema' + +/** + * Small releases the settler just settled (at most 3 files, under 1 MB), queued for the merger's sidecar fold + * (app/services/sidecar_folder.ts): an .nfo, .sfv or .nzb that came apart from its upload (another counter, another + * name) joins it. A queue, so the fold needn't scan the releases (LOG.md 2026-10-15 (57)). + */ +export default class extends BaseSchema { + async up() { + this.schema.raw(` + CREATE TABLE fold_candidates ( + release_id bigint PRIMARY KEY, + created_at timestamptz NOT NULL DEFAULT now() + )`) + } + + async down() { + this.schema.raw(`DROP TABLE IF EXISTS fold_candidates`) + } +} diff --git a/tests/functional/par2_merger.spec.ts b/tests/functional/par2_merger.spec.ts index 168dd92..4c40086 100644 --- a/tests/functional/par2_merger.spec.ts +++ b/tests/functional/par2_merger.spec.ts @@ -2,6 +2,8 @@ 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' +import { encodeChunk } from '#lib/chunks/index' +import ChunkPartitions from '#services/chunk_partitions' const hash = (c: string) => c.repeat(32) @@ -47,7 +49,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, stems: 0, stemMerges: 0, stemReleases: 0 }) + assert.deepEqual(round, { sets: 1, merges: 1, releases: 3, stems: 0, stemMerges: 0, stemReleases: 0, folded: { looked: 0, sidecars: 0, folded: 0 } }) const target = await db.from('releases').where('id', b).firstOrFail() assert.equal(target.files, 4) assert.equal(target.segments, 40) @@ -180,4 +182,40 @@ test.group('Par2Merger', (group) => { assert.equal((await new Par2Merger().round()).merges, 0) assert.equal((await db.from('releases').whereIn('id', ids).select('id')).length, 3) }) + + test('an .nzb or .nfo apart from its upload is folded into it', async ({ assert }) => { + await ChunkPartitions.ensure() + const at = new Date('2022-07-08T03:46:00Z') + const poster = 'up ' + let n = 0 + const release = async (name: string, files: number, total: number | null, posted: Date, chunkFiles: string[]) => { + n++ + const key = `ff${String(n).padStart(6, '0')}`.padEnd(40, '0') + const [row] = await db + .table('releases') + .insert({ release_key: key, identity_key: key, name, source: 'stem', groups: '[]', group_id: 1, files, file_total: total, poster, posted_at: posted, readable: true, settled_at: posted }) + .returning('id') + const id = Number(row.id) + const payload = encodeChunk(chunkFiles.map((f, i) => ({ name: f, poster, subject: `${name} [${String(i).padStart(2, '0')}/${total ?? 0}] - "${f}" yEnc`, postedAt: posted.getTime(), groups: ['a.b.test'], segments: [[1, `${id}-${i}@x`, 1000]] }))) + await db.table('release_chunks').insert({ release_id: id, range_id: 1, batch_from: 1, files: chunkFiles.length, segments: chunkFiles.length, payload }) + await db.table('fold_candidates').insert({ release_id: id }).onConflict('release_id').ignore() + return id + } + const upload = await release('Murdoch Mysteries S8D1', 94, 94, at, ['Murdoch Mysteries S8D1.part01.rar', 'Murdoch Mysteries S8D1.part02.rar', 'Murdoch Mysteries S8D1.part03.rar', 'Murdoch Mysteries S8D1.part04.rar']) + const other = await release('Another Show S1D1', 50, 50, new Date(at.getTime() + 120_000), ['Another Show S1D1.part01.rar', 'Another Show S1D1.part02.rar', 'Another Show S1D1.part03.rar', 'Another Show S1D1.part04.rar']) + const nzb = await release('Murdoch Mysteries S8D1', 1, 80, new Date(at.getTime() + 60_000), ['Murdoch Mysteries S8D1.nzb']) + // A picture isn't a sidecar; an nfo an hour later belongs to nothing here. + const picture = await release('cover', 1, null, at, ['cover.jpg']) + const late = await release('Murdoch Mysteries S8D1', 1, null, new Date(at.getTime() + 3600_000), ['Murdoch Mysteries S8D1.nfo']) + + const round = await new Par2Merger().round() + assert.deepEqual(round.folded, { looked: 5, sidecars: 2, folded: 1 }) + assert.isNull(await db.from('releases').where('id', nzb).first()) + assert.equal((await db.from('releases').where('id', upload).firstOrFail()).files, 95) + assert.equal((await db.from('releases').where('id', other).firstOrFail()).files, 50) + assert.isNotNull(await db.from('releases').where('id', picture).first()) + assert.isNotNull(await db.from('releases').where('id', late).first()) + assert.equal((await db.from('release_merges').where({ into_id: upload, via: 'sidecar' }).count('* as c'))[0].c, '1') + assert.equal((await db.from('fold_candidates').count('* as c'))[0].c, '0') + }) })