Files
usenet-indexer/app/services/par2_merger.ts
T
Daan MeijerandClaude Opus 5.5 ebe9f8bb36 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 <noreply@anthropic.com>
2026-10-23 12:00:19 +02:00

394 lines
21 KiB
TypeScript
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
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'
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
/**
* A set held by more releases nearby than this is one posted again and again, not one upload's par2 (every
* "kantodoNNN.bin" test upload carries the same set: 64 releases became one of 631 files; LOG.md 2026-10-12 (30)).
*/
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 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 end of its last whole window, kept over a restart. */
const CURSOR = 'merger'
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
/** The sidecar fold: candidates looked at, sidecar-only ones among them, folded into their upload. */
folded?: { looked: number; sidecars: number; folded: 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})\\.')`
/** 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
group_id: number
files: number
posted_at: Date | null
readable: boolean
poster?: string | null
file_total?: number | null
}
/**
* The par2 merge (REBUILD.md, phase 2): releases whose files a par2 set
* describes are one upload. Obfuscated uploads post each file under a name of
* its own ("[9/20] - "14f5999a7bd4680b0720e524cac0b3b5"", hep: 13,232
* one-file releases), so the header pass can't group them; the prober reads
* each such file's first article (a `lone` candidate) and stores the MD5 of
* its first 16 KiB, and a par2 body's file list holds the same hash per file,
* whatever the file was renamed to. Releases in one group, posted within
* MAX_SPREAD_MS of a release that holds the set's par2, become the one with
* 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
private folder = new SidecarFolder((trx, found, key, via) => this.merge(trx, found, key, via))
/** Null until read from worker_cursors (1970 the first time: everything). */
private since: Date | null = null
stop() {
this.stopped = true
}
async run() {
let prunedAt = 0
while (!this.stopped) {
const started = Date.now()
try {
// First, at start too: a prune needn't wait for a long round.
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, ${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).`)
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}`)
}
for (let waited = Date.now() - started; waited < ROUND_MS && !this.stopped; waited += 1000) await new Promise((r) => setTimeout(r, 1000))
}
}
/**
* 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<MergeRound> {
// 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)
}
// 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()
let foldedAt = Date.now()
const foldAgain = async () => {
if (Date.now() - foldedAt < FOLD_EVERY_MS) return
const more = await this.folder.round()
folded.looked += more.looked
folded.sidecars += more.sidecars
folded.folded += more.folded
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 round: MergeRound = { sets: 0, merges: 0, releases: 0, stems: 0, stemMerges: 0, stemReleases: 0, folded }
while (this.since < end) {
if (this.stopped) return round
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]
)
}
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<Date | null> {
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<number> {
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<string, Member>()
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<number> {
return db.transaction(async (trx) => {
const { rows: holders } = await trx.rawQuery(
`SELECT r.id, r.release_key, r.group_id, r.files, r.posted_at, r.readable FROM par2_sets s JOIN releases r ON r.id = s.release_id
WHERE s.set_id = ? ORDER BY r.id`,
[setId]
)
const anchor: Member | undefined = holders[0]
if (!anchor) return 0
const at = (r: Member) => (r.posted_at ? new Date(r.posted_at).getTime() : null)
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 CROSS JOIN LATERAL (${HOLDERS}) h JOIN releases r ON r.id = h.release_id
WHERE f.set_id = ?`,
[setId]
)
// A file of the set that matches more than one release nearby is ambiguous: files that share their first
// 16 KiB and size (131 uploads of 1 GiB test files "kantodoNNN.bin" matched one set's single file and became
// one release of 1,284 files; LOG.md 2026-10-12 (16)). A plain repost of the same file stays apart too.
const byFile = new Map<string, Map<string, Member>>()
for (const m of matches.filter(near)) {
const releases = byFile.get(m.file_id) ?? new Map<string, Member>()
releases.set(String(m.id), m)
byFile.set(m.file_id, releases)
}
const nearHolders = holders.filter(near)
if (nearHolders.length > MAX_HOLDERS) return 0
const found = new Map<string, Member>()
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<string, Member>, key: string, via: 'par2' | 'inner' | 'sidecar'): Promise<number> {
{
// 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 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 = 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, 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])
}
// 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]
)
// The members' 7z tails together: those below the highest volume go.
await pruneSevenZipTails(trx, [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(
`DELETE FROM par2_sets p WHERE p.release_id = ANY(?)
AND (EXISTS (SELECT 1 FROM par2_sets t WHERE t.set_id = p.set_id AND t.release_id = ?)
OR EXISTS (SELECT 1 FROM par2_sets o WHERE o.set_id = p.set_id AND o.release_id = ANY(?) AND o.release_id < p.release_id))`,
[others, target.id, others]
)
await trx.rawQuery(`UPDATE par2_sets SET release_id = ? WHERE release_id = ANY(?)`, [target.id, others])
await trx.rawQuery(
`UPDATE releases t SET
files = t.files + s.files, segments = t.segments + s.segments, bytes = t.bytes + s.bytes, chunks = t.chunks + s.chunks,
first_article = least(t.first_article, s.first_article), last_article = greatest(t.last_article, s.last_article),
posted_at = least(t.posted_at, s.posted_at), last_posted_at = greatest(t.last_posted_at, s.last_posted_at),
first_seen_at = least(t.first_seen_at, s.first_seen_at), last_seen_at = greatest(t.last_seen_at, s.last_seen_at),
readable = t.readable OR s.readable, has_par2 = t.has_par2 OR s.has_par2, passworded = t.passworded OR s.passworded,
file_total = greatest(t.file_total, s.file_total), inner_name = coalesce(t.inner_name, s.inner_name),
name = coalesce(nullif(?, ''), t.name)
FROM (SELECT sum(files) AS files, sum(segments) AS segments, sum(bytes) AS bytes, sum(chunks) AS chunks,
min(first_article) AS first_article, max(last_article) AS last_article, min(posted_at) AS posted_at,
max(last_posted_at) AS last_posted_at, min(first_seen_at) AS first_seen_at, max(last_seen_at) AS last_seen_at,
bool_or(readable) AS readable, bool_or(has_par2) AS has_par2, bool_or(passworded) AS passworded,
max(file_total) AS file_total, max(inner_name) AS inner_name
FROM releases WHERE id = ANY(?)) s
WHERE t.id = ?`,
[name ?? '', others, target.id]
)
// 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 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
}
}
/** The name a set's files share (par2ReleaseName), for an upload none of whose names is readable. */
private async setName(trx: any, setId: string): Promise<string | undefined> {
const { rows } = await trx.rawQuery(`SELECT name FROM par2_files WHERE set_id = ?`, [setId])
const name = par2ReleaseName(rows.map((r: { name: string }) => r.name))
// A par2 made after the renaming names the random names again.
return name && !looksRandom(name) ? name.slice(0, 255) : undefined
}
}