Judge the releases whose facts changed every 2 minutes, all daily
A full round took 35-38 min of each hour and grew with every inspection the prober added. Triggers note changed releases in judge_dirty; the judge judges those (and releases a file hash of theirs tipped to repeated) by every rule, and every release in a daily full round for what time changes. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
1 parent
036e2a8397
commit
76e52f69ce
6 files changed
+366
-63
No files matched your search
+164
-53
@@ -1,10 +1,18 @@
|
||||
import db from '@adonisjs/lucid/services/db'
|
||||
import logger from '@adonisjs/core/services/logger'
|
||||
|
||||
/** Raised when a rule changes: verdicts of an older version are judged again. */
|
||||
/** Raised when a rule changes: verdicts of an older version are judged again (a full round, right away). */
|
||||
export const JUDGE_VERSION = 1
|
||||
/** Between rounds: each rule is a scan of its table. */
|
||||
const ROUND_MS = 60 * 60 * 1000
|
||||
/** Between rounds of the changed releases (judge_dirty, noted by triggers; migration 4900). */
|
||||
const ROUND_MS = 2 * 60 * 1000
|
||||
/**
|
||||
* A full round (every release, each rule a scan of its table: 35–38 min) this often: for what no trigger notes. Time
|
||||
* (unusable a day after settling, a release final once the ranges beside it are read) and fewer releases with a file
|
||||
* after a merge (repeated).
|
||||
*/
|
||||
const FULL_MS = 24 * 60 * 60 * 1000
|
||||
/** Changed releases judged at a time. */
|
||||
const CHUNK = 50_000
|
||||
/** A file's first 16 KiB and size in this many releases is one file posted again and again (test files, spam). */
|
||||
const REPEATED_RELEASES = 20
|
||||
|
||||
@@ -24,10 +32,18 @@ export interface Rule {
|
||||
name: string
|
||||
/** Why, in a few words (the log, the web interface). */
|
||||
says: string
|
||||
/** SELECT of the release ids the rule judges. */
|
||||
sql: string
|
||||
/** SELECT of the release ids the rule judges; scoped, only those in judge_scope. */
|
||||
sql: (scoped: boolean) => string
|
||||
}
|
||||
|
||||
/** A join to the releases in scope (an incremental round), on the given release id column. */
|
||||
const only = (scoped: boolean, column: string) => (scoped ? `JOIN judge_scope s ON s.release_id = ${column}` : '')
|
||||
|
||||
/** The file hashes (first 16 KiB and size) of the releases in scope that are in REPEATED_RELEASES+ releases. */
|
||||
const TIPPED = `hashes AS (SELECT DISTINCT h.hash16k, h.size FROM file_hashes h JOIN judge_scope s ON s.release_id = h.release_id),
|
||||
tipped AS (SELECT x.hash16k, x.size FROM hashes x
|
||||
WHERE (SELECT count(DISTINCT f.release_id) FROM file_hashes f WHERE f.hash16k = x.hash16k AND f.size = x.size) >= ${REPEATED_RELEASES})`
|
||||
|
||||
/**
|
||||
* The junk judge's rules, in order: a release takes the first that matches (REBUILD.md, junk). Each runs on what
|
||||
* the header pass, the prober, the merges and the spots know by now.
|
||||
@@ -36,26 +52,33 @@ export const RULES: Rule[] = [
|
||||
{
|
||||
name: 'dead',
|
||||
says: 'every probed first article is gone from the server (taken down or expired)',
|
||||
sql: `SELECT release_id FROM probe_candidates WHERE status IN ('done', 'missing', 'error') GROUP BY 1 HAVING bool_and(status = 'missing')`,
|
||||
sql: (scoped) =>
|
||||
`SELECT c.release_id FROM probe_candidates c ${only(scoped, 'c.release_id')}
|
||||
WHERE c.status IN ('done', 'missing', 'error') GROUP BY 1 HAVING bool_and(c.status = 'missing')`,
|
||||
},
|
||||
{
|
||||
name: 'fake',
|
||||
says: 'named as a film, episode or music, but the content is a program',
|
||||
// An archive whose entries are all executables, under a media name (its own or its entries'); or a file that is
|
||||
// an executable by name or by its first bytes, under a media name.
|
||||
sql: `SELECT i.release_id FROM release_inspections i JOIN releases r ON r.id = i.release_id
|
||||
sql: (scoped) => `SELECT i.release_id FROM release_inspections i JOIN releases r ON r.id = i.release_id ${only(scoped, 'i.release_id')}
|
||||
WHERE i.kind = 'archive' AND jsonb_array_length(i.result->'names') > 0
|
||||
AND NOT EXISTS (SELECT 1 FROM jsonb_array_elements_text(i.result->'names') n WHERE n !~* '${EXECUTABLE}')
|
||||
AND (r.name ~* '${MEDIA}' OR (i.result->>'names') ~* '${MEDIA}')
|
||||
UNION
|
||||
SELECT i.release_id FROM release_inspections i JOIN releases r ON r.id = i.release_id
|
||||
SELECT i.release_id FROM release_inspections i JOIN releases r ON r.id = i.release_id ${only(scoped, 'i.release_id')}
|
||||
WHERE i.kind = 'lone' AND ((i.result->>'yencName') ~* '${EXECUTABLE}' OR i.result->>'type' = 'exe')
|
||||
AND (r.name ~* '${MEDIA}' OR (i.result->>'yencName') ~* '${MEDIA}')`,
|
||||
},
|
||||
{
|
||||
name: 'repeated',
|
||||
says: `the same file content in ${REPEATED_RELEASES}+ releases (test files, spam)`,
|
||||
sql: `SELECT DISTINCT h.release_id FROM file_hashes h
|
||||
// Scoped: the hashes of the releases in scope, counted each (a round first adds the other releases a hash of theirs
|
||||
// tipped over: ReleaseJudge.changed).
|
||||
sql: (scoped) =>
|
||||
scoped
|
||||
? `WITH ${TIPPED} SELECT DISTINCT h.release_id FROM file_hashes h JOIN judge_scope s ON s.release_id = h.release_id JOIN tipped USING (hash16k, size)`
|
||||
: `SELECT DISTINCT h.release_id FROM file_hashes h
|
||||
JOIN (SELECT hash16k, size FROM file_hashes GROUP BY 1, 2 HAVING count(DISTINCT release_id) >= ${REPEATED_RELEASES}) x USING (hash16k, size)`,
|
||||
},
|
||||
{
|
||||
@@ -66,20 +89,24 @@ export const RULES: Rule[] = [
|
||||
// titles (LOG.md 2026-10-16 (64)). Not Celeborn, CelebTA or other posters with "celeb" in their name.
|
||||
// By the posters table's trigram index, each poster's releases then by releases_poster_newest_idx: 7 s, where the
|
||||
// regex on all of releases was 45 s, every round (LOG.md 2026-10-19 (80)).
|
||||
sql: `SELECT r.id FROM posters p JOIN releases r ON left(r.poster, 200) = p.poster
|
||||
sql: (scoped) =>
|
||||
scoped
|
||||
? `SELECT r.id FROM releases r ${only(scoped, 'r.id')}
|
||||
WHERE r.poster ~* '${SCRAPED_POSTERS}' AND r.source = 'subject' AND r.file_total IS NULL`
|
||||
: `SELECT r.id FROM posters p JOIN releases r ON left(r.poster, 200) = p.poster
|
||||
WHERE p.poster ~* '${SCRAPED_POSTERS}' AND r.poster ~* '${SCRAPED_POSTERS}' AND r.source = 'subject' AND r.file_total IS NULL`,
|
||||
},
|
||||
{
|
||||
name: 'locked',
|
||||
says: 'passworded, and no spot, NZB or title gives the password',
|
||||
// releases_passworded_idx: a scan of all of releases was 30 s a round.
|
||||
sql: `SELECT r.id FROM releases r WHERE r.passworded AND NOT EXISTS (SELECT 1 FROM release_passwords p WHERE p.release_id = r.id)`,
|
||||
sql: (scoped) => `SELECT r.id FROM releases r ${only(scoped, 'r.id')} WHERE r.passworded AND NOT EXISTS (SELECT 1 FROM release_passwords p WHERE p.release_id = r.id)`,
|
||||
},
|
||||
{
|
||||
name: 'unusable',
|
||||
says: 'random names, no name inside, no par2 or inner-name merge, no spot, a day after settling, and final',
|
||||
// Final (release_final): next to an unread range its par2 or the rest of its files may still come (LOG.md 2026-10-16 (67)).
|
||||
sql: `SELECT r.id FROM releases r
|
||||
sql: (scoped) => `SELECT r.id FROM releases r ${only(scoped, 'r.id')}
|
||||
WHERE r.settled_at < now() - interval '1 day' AND NOT r.readable AND r.inner_name IS NULL
|
||||
AND release_final(r.group_id, r.first_article, r.last_article, r.posted_at, r.last_posted_at, r.complete)
|
||||
AND NOT EXISTS (SELECT 1 FROM release_merges m WHERE m.into_id = r.id)
|
||||
@@ -89,15 +116,18 @@ export const RULES: Rule[] = [
|
||||
]
|
||||
|
||||
export interface JudgeRound {
|
||||
/** Per rule: releases it judges now, verdicts added, verdicts taken back. */
|
||||
/** Releases judged: every one (a full round), or the changed ones and those a hash of theirs tipped over. */
|
||||
releases?: number
|
||||
/** Per rule: releases it judges now (in scope), verdicts it gained, verdicts it lost. */
|
||||
rules: Record<string, { matched: number; added: number; removed: number }>
|
||||
}
|
||||
|
||||
/**
|
||||
* The junk judge (REBUILD.md, junk; voyager 1's second filter, D42): after the header pass's rules (garbage
|
||||
* signatures, executables, …) a release is judged again once it's probed, by what its articles showed. Each
|
||||
* round recomputes every rule: a release a rule matches gets a verdict (the first rule in order, over a later one's), one it no longer
|
||||
* matches loses its verdict. Verdicts only hide (search leaves them out); nothing is dropped yet.
|
||||
* signatures, executables, …) a release is judged again once it's probed, by what its articles showed. Every few
|
||||
* minutes the releases whose facts changed (judge_dirty) are judged by every rule; a full round judges every release
|
||||
* daily. A release gets the verdict of the first rule in order that matches it; one that no longer matches loses its
|
||||
* verdict. Verdicts only hide (search leaves them out); nothing is dropped yet.
|
||||
*/
|
||||
export class ReleaseJudge {
|
||||
private stopped = false
|
||||
@@ -110,9 +140,13 @@ export class ReleaseJudge {
|
||||
while (!this.stopped) {
|
||||
const started = Date.now()
|
||||
try {
|
||||
const round = await this.round()
|
||||
const parts = Object.entries(round.rules).map(([n, r]) => `${n} ${r.matched} (+${r.added} −${r.removed})`)
|
||||
logger.info(`Judged: ${parts.join(', ')}.`)
|
||||
if (await this.fullDue()) {
|
||||
const round = await this.full()
|
||||
if (!this.stopped) logger.info(`Judged all releases: ${describe(round)}.`)
|
||||
} else {
|
||||
const round = await this.changed()
|
||||
if (round.releases! > 0) logger.info(`Judged ${round.releases!.toLocaleString('en')} changed releases: ${describe(round)}.`)
|
||||
}
|
||||
} catch (error: any) {
|
||||
logger.error(`Judge round failed: ${error.message}`)
|
||||
}
|
||||
@@ -120,47 +154,124 @@ export class ReleaseJudge {
|
||||
}
|
||||
}
|
||||
|
||||
/** One round over every rule; with dry, only counts. */
|
||||
async round(dry = false): Promise<JudgeRound> {
|
||||
/** A full round is due: a day after the last, or the last was by another rule set. */
|
||||
async fullDue(): Promise<boolean> {
|
||||
const state = await db.from('judge_state').first()
|
||||
return !state || state.version !== JUDGE_VERSION || Date.now() - new Date(state.full_at).getTime() >= FULL_MS
|
||||
}
|
||||
|
||||
/**
|
||||
* Every release by every rule, the matches gathered in judge_full (each rule its own statement: no transaction of
|
||||
* half an hour), then applied at once. With dry, only counts. Stopped halfway, nothing is applied.
|
||||
*/
|
||||
async full(dry = false): Promise<JudgeRound> {
|
||||
const started = new Date()
|
||||
const round: JudgeRound = { rules: {} }
|
||||
// Verdicts of an older rule set are judged again from scratch.
|
||||
if (!dry) await db.from('release_verdicts').where('rule_version', '<>', JUDGE_VERSION).delete()
|
||||
const judged = new Set<string>()
|
||||
await db.rawQuery(`TRUNCATE judge_full`)
|
||||
for (const rule of RULES) {
|
||||
if (this.stopped) break
|
||||
const { rows } = await db.rawQuery(rule.sql)
|
||||
const ids = rows.map((r: Record<string, unknown>) => String(Object.values(r)[0])).filter((id: string) => !judged.has(id))
|
||||
for (const id of ids) judged.add(id)
|
||||
const stats = { matched: ids.length, added: 0, removed: 0 }
|
||||
if (!dry) {
|
||||
await db.transaction(async (trx) => {
|
||||
// Dropped first: inside an outer transaction (the tests) ON COMMIT DROP comes only at its end.
|
||||
await trx.rawQuery(`DROP TABLE IF EXISTS judged`)
|
||||
await trx.rawQuery(`CREATE TEMP TABLE judged (release_id bigint PRIMARY KEY) ON COMMIT DROP`)
|
||||
for (let i = 0; i < ids.length; i += 50_000) {
|
||||
await trx.rawQuery(`INSERT INTO judged SELECT unnest(?::bigint[]) ON CONFLICT DO NOTHING`, [ids.slice(i, i + 50_000)])
|
||||
}
|
||||
const removed = await trx.rawQuery(
|
||||
`DELETE FROM release_verdicts v WHERE v.rule = ? AND NOT EXISTS (SELECT 1 FROM judged j WHERE j.release_id = v.release_id)`,
|
||||
if (this.stopped) return round
|
||||
const { rowCount } = await db.rawQuery(
|
||||
`INSERT INTO judge_full (release_id, rule) SELECT id, ? FROM (${rule.sql(false)}) AS x(id) ON CONFLICT DO NOTHING`,
|
||||
[rule.name]
|
||||
)
|
||||
round.rules[rule.name] = { matched: rowCount ?? 0, added: 0, removed: 0 }
|
||||
}
|
||||
if (dry || this.stopped) return round
|
||||
await db.rawQuery(`ANALYZE judge_full`)
|
||||
await db.transaction((trx) => this.apply(trx, 'judge_full', false, round))
|
||||
await db.rawQuery(
|
||||
`INSERT INTO judge_state (id, full_at, version) VALUES (1, ?, ?) ON CONFLICT (id) DO UPDATE SET full_at = excluded.full_at, version = excluded.version`,
|
||||
[started, JUDGE_VERSION]
|
||||
)
|
||||
return round
|
||||
}
|
||||
|
||||
/**
|
||||
* The releases noted in judge_dirty since, CHUNK at a time, each chunk in a transaction (a failed one stays noted).
|
||||
* A changed release can bring a file to REPEATED_RELEASES releases: the others with it are judged too (all but those
|
||||
* already repeated, or dead or fake, which come first).
|
||||
*/
|
||||
async changed(): Promise<JudgeRound> {
|
||||
const round: JudgeRound = { releases: 0, rules: {} }
|
||||
for (const rule of RULES) round.rules[rule.name] = { matched: 0, added: 0, removed: 0 }
|
||||
let more = true
|
||||
while (more && !this.stopped) {
|
||||
await db.transaction(async (trx) => {
|
||||
// Dropped first: inside an outer transaction (the tests) ON COMMIT DROP comes only at its end.
|
||||
await trx.rawQuery(`DROP TABLE IF EXISTS judge_scope`)
|
||||
await trx.rawQuery(`DROP TABLE IF EXISTS judged`)
|
||||
await trx.rawQuery(`CREATE TEMP TABLE judge_scope (release_id bigint PRIMARY KEY) ON COMMIT DROP`)
|
||||
const { rows } = await trx.rawQuery(
|
||||
`WITH taken AS (DELETE FROM judge_dirty WHERE ctid = ANY(ARRAY(SELECT ctid FROM judge_dirty LIMIT ${CHUNK})) RETURNING release_id),
|
||||
scoped AS (INSERT INTO judge_scope SELECT DISTINCT release_id FROM taken ON CONFLICT DO NOTHING RETURNING 1)
|
||||
SELECT (SELECT count(*) FROM taken) AS taken`
|
||||
)
|
||||
const taken = Number(rows[0].taken)
|
||||
more = taken === CHUNK
|
||||
if (taken === 0) return
|
||||
await trx.rawQuery(
|
||||
`WITH ${TIPPED}
|
||||
INSERT INTO judge_scope SELECT DISTINCT f.release_id FROM file_hashes f JOIN tipped USING (hash16k, size)
|
||||
WHERE NOT EXISTS (SELECT 1 FROM release_verdicts v WHERE v.release_id = f.release_id AND v.rule IN ('dead', 'fake', 'repeated'))
|
||||
ON CONFLICT DO NOTHING`
|
||||
)
|
||||
await trx.rawQuery(`ANALYZE judge_scope`)
|
||||
const { rows: scope } = await trx.rawQuery(`SELECT count(*) AS n FROM judge_scope`)
|
||||
round.releases! += Number(scope[0].n)
|
||||
await trx.rawQuery(`CREATE TEMP TABLE judged (release_id bigint PRIMARY KEY, rule varchar(16) NOT NULL) ON COMMIT DROP`)
|
||||
for (const rule of RULES) {
|
||||
const { rowCount } = await trx.rawQuery(
|
||||
`INSERT INTO judged (release_id, rule) SELECT id, ? FROM (${rule.sql(true)}) AS x(id) ON CONFLICT DO NOTHING`,
|
||||
[rule.name]
|
||||
)
|
||||
// Only the releases whose verdict changes: each row through the insert's conflict check took 60–150 s a rule
|
||||
// for a few thousand changes in a million matches (LOG.md 2026-10-19 (80)).
|
||||
await trx.rawQuery(`ANALYZE judged`)
|
||||
const added = await trx.rawQuery(
|
||||
`INSERT INTO release_verdicts (release_id, rule, rule_version)
|
||||
SELECT j.release_id, ?, ? FROM judged j LEFT JOIN release_verdicts v ON v.release_id = j.release_id
|
||||
WHERE v.rule IS DISTINCT FROM ? AND EXISTS (SELECT 1 FROM releases r WHERE r.id = j.release_id)
|
||||
ON CONFLICT (release_id) DO UPDATE SET rule = excluded.rule, rule_version = excluded.rule_version
|
||||
WHERE release_verdicts.rule <> excluded.rule`,
|
||||
[rule.name, JUDGE_VERSION, rule.name]
|
||||
)
|
||||
stats.removed = removed.rowCount ?? 0
|
||||
stats.added = added.rowCount ?? 0
|
||||
})
|
||||
}
|
||||
round.rules[rule.name] = stats
|
||||
round.rules[rule.name].matched += rowCount ?? 0
|
||||
}
|
||||
await trx.rawQuery(`ANALYZE judged`)
|
||||
await this.apply(trx, 'judged', true, round)
|
||||
})
|
||||
}
|
||||
return round
|
||||
}
|
||||
|
||||
/**
|
||||
* The matches (judged: release_id, rule; the first rule's each) into release_verdicts: verdicts no rule gives any
|
||||
* more removed (scoped: of the releases in judge_scope only), changed ones written. Only the changes go through the
|
||||
* insert: each match through its conflict check took 60–150 s a rule (LOG.md 2026-10-19 (80)).
|
||||
*/
|
||||
private async apply(trx: any, judged: string, scoped: boolean, round: JudgeRound) {
|
||||
const tally = (rows: { rule: string; n: string }[], field: 'added' | 'removed') => {
|
||||
for (const { rule, n } of rows) {
|
||||
round.rules[rule] ??= { matched: 0, added: 0, removed: 0 }
|
||||
round.rules[rule][field] += Number(n)
|
||||
}
|
||||
}
|
||||
const { rows: removed } = await trx.rawQuery(
|
||||
`WITH gone AS (DELETE FROM release_verdicts v
|
||||
WHERE ${scoped ? 'v.release_id IN (SELECT release_id FROM judge_scope) AND' : ''}
|
||||
NOT EXISTS (SELECT 1 FROM ${judged} j WHERE j.release_id = v.release_id)
|
||||
RETURNING v.rule)
|
||||
SELECT rule, count(*) AS n FROM gone GROUP BY 1`
|
||||
)
|
||||
tally(removed, 'removed')
|
||||
const { rows: added } = await trx.rawQuery(
|
||||
`WITH written AS (INSERT INTO release_verdicts (release_id, rule, rule_version)
|
||||
SELECT j.release_id, j.rule, ? FROM ${judged} j LEFT JOIN release_verdicts v ON v.release_id = j.release_id
|
||||
WHERE v.rule IS DISTINCT FROM j.rule AND EXISTS (SELECT 1 FROM releases r WHERE r.id = j.release_id)
|
||||
ON CONFLICT (release_id) DO UPDATE SET rule = excluded.rule, rule_version = excluded.rule_version
|
||||
WHERE release_verdicts.rule <> excluded.rule
|
||||
RETURNING rule)
|
||||
SELECT rule, count(*) AS n FROM written GROUP BY 1`,
|
||||
[JUDGE_VERSION]
|
||||
)
|
||||
tally(added, 'added')
|
||||
}
|
||||
}
|
||||
|
||||
/** "dead 1,157,018 (+1,136 −712), …" for the log. */
|
||||
function describe(round: JudgeRound): string {
|
||||
return Object.entries(round.rules)
|
||||
.map(([name, r]) => `${name} ${r.matched.toLocaleString('en')} (+${r.added.toLocaleString('en')} −${r.removed.toLocaleString('en')})`)
|
||||
.join(', ')
|
||||
}
|
||||
Reference in new issue
Block a user