Daan MeijerandClaude Opus 5.5
2823d42bfa
hep, hephaestus and gyges post each file under a subject name of its own
(31,000 one-file releases), but the yEnc name in each file's first article
keeps the upload's random stem ("bX147zx8I0oEgj7nzFCiorCT0ps2jEcb.7z.001"),
one stem per counter total. The merger's second pass joins releases whose
lone inspections share a random stem: one group and poster, the same total,
within 48 h, no more files than the total plus 10. An expression index over
the stem; release_merges records the kind (via) and takes a stem.
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
273 lines
14 KiB
TypeScript
273 lines
14 KiB
TypeScript
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'
|
||
|
||
/** 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
|
||
/** Between rounds. */
|
||
const ROUND_MS = 60_000
|
||
/** A round looks back this far past the previous one's start (rows committed while it ran). */
|
||
const OVERLAP_MS = 60_000
|
||
|
||
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
|
||
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 since = new Date(0)
|
||
|
||
stop() {
|
||
this.stopped = true
|
||
}
|
||
|
||
async run() {
|
||
while (!this.stopped) {
|
||
const started = Date.now()
|
||
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}`)
|
||
}
|
||
for (let waited = Date.now() - started; waited < ROUND_MS && !this.stopped; waited += 1000) await new Promise((r) => setTimeout(r, 1000))
|
||
}
|
||
}
|
||
|
||
/** The sets with something new since the last round: a par2 found, or a hash of one of their files. */
|
||
async round(): Promise<MergeRound> {
|
||
const { rows: now } = await db.rawQuery(`SELECT now() AS at`)
|
||
const since = new Date(this.since.getTime() - OVERLAP_MS)
|
||
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 }
|
||
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).
|
||
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 ${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<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 JOIN file_hashes h ON h.hash16k = f.hash16k AND h.size = f.length 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'): 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])
|
||
}
|
||
// 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.
|
||
await trx.rawQuery(
|
||
`UPDATE releases SET complete = CASE WHEN file_total IS NULL THEN NULL 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])
|
||
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
|
||
}
|
||
}
|