par2 file lists dropped once their holders are final; late candidates claimable
Most par2 sets describe readable uploads whose files are never hashed, so they were never complete and kept forever (19 GB). Candidates added to a release after it settled were never marked ready (93,642 since 2026-10-13); the batch store and merge mark them now, migration 5100 repairs the waiting ones (LOG.md 2026-10-20 (87)). Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
1 parent
4f41303482
commit
8ad5ea162c
7 files changed
+152
-61
No files matched your search
@@ -242,6 +242,10 @@ It reads what's new an hour (`created_at`) at a time, its place kept after
|
||||
each, and merges only the sets held or matched by two releases or more; a
|
||||
file whose first 16 KiB and size more than 20 files share is a common one and
|
||||
matches nothing (LOG.md 2026-10-20 (85)).
|
||||
A set's file list is dropped once it can't merge anything more: complete (all
|
||||
its files in its holder), judged (every holder junk), or final (every holder
|
||||
inside a run of done ranges, those ranges' candidates probed, its hashes seen
|
||||
by the merger); `par2_sets` stays (LOG.md 2026-10-20 (87)).
|
||||
|
||||
**Backlog: voyager 1's features not in voyager 2** (CONCEPT.md decisions;
|
||||
voyager 1's data exported to `/mnt/bulk/usenet-rebuild/v1-export/` on
|
||||
|
||||
@@ -316,6 +316,15 @@ class BatchStore {
|
||||
.ignore()
|
||||
.returning('id')
|
||||
out.candidates += added.length
|
||||
// Files added to a settled release (a later range of its upload): the settler made its candidates claimable
|
||||
// already and won't again (93,642 waited since 2026-10-13; LOG.md 2026-10-20 (87)).
|
||||
if (added.length > 0) {
|
||||
await trx.rawQuery(
|
||||
`UPDATE probe_candidates pc SET ready = true, unreadable = NOT r.readable FROM releases r
|
||||
WHERE pc.id = ANY(?) AND r.id = pc.release_id AND r.settled_at IS NOT NULL`,
|
||||
[added.map((a: { id: string | number }) => String(a.id))]
|
||||
)
|
||||
}
|
||||
}
|
||||
|
||||
return out
|
||||
|
||||
@@ -115,7 +115,7 @@ export class Par2Merger {
|
||||
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.`)
|
||||
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).`)
|
||||
@@ -335,6 +335,12 @@ export class Par2Merger {
|
||||
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]
|
||||
)
|
||||
// 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(
|
||||
|
||||
+62
-50
@@ -4,86 +4,98 @@ import db from '@adonisjs/lucid/services/db'
|
||||
const SETTLED_MS = 2 * 86400_000
|
||||
/** Sets per statement. */
|
||||
const BATCH = 20_000
|
||||
/** The complete pass's row in worker_cursors: the newest holder's created_at of the last set it looked at. */
|
||||
/** The sweep's row in worker_cursors: the last set_id it looked at (key); past the last set it starts over. */
|
||||
const CURSOR = 'par2-prune'
|
||||
/** The merger's row in worker_cursors: the end of its last whole window. */
|
||||
const MERGER = 'merger'
|
||||
/** The merger's windows look back this far (Par2Merger's OVERLAP_MS). */
|
||||
const MERGER_OVERLAP_MS = 60_000
|
||||
/** Verdicts whose releases needn't merge any more; locked isn't one (a password found takes it back). */
|
||||
const JUNK = ['dead', 'fake', 'repeated']
|
||||
/** Probe candidates whose results a merge reads: a file's hash (lone), a set's file list (par2). */
|
||||
const MERGE_KINDS = ['lone', 'par2']
|
||||
|
||||
export interface PruneRound {
|
||||
complete: number
|
||||
judged: number
|
||||
final: number
|
||||
rows: number
|
||||
}
|
||||
|
||||
/**
|
||||
* Drops par2_files rows the merge can't use any more (LOG.md 2026-10-16 (60)): the table only grew, 35M rows and
|
||||
* 13.6 GB at 1.8M sets. A set's file list matches file hashes to releases and names an unreadable upload once; it's
|
||||
* done when
|
||||
* Drops par2_files rows the merge can't use any more (LOG.md 2026-10-16 (60), 2026-10-20 (87)). A set's file list
|
||||
* matches file hashes to releases and names an unreadable upload once; it's done when
|
||||
* - complete: SETTLED_MS after its last holder's par2, every file it describes has its hash in a release holding the
|
||||
* set (all of it merged into one);
|
||||
* - judged: every release holding it has a junk verdict (dead, fake, repeated).
|
||||
* - judged: every release holding it has a junk verdict (dead, fake, repeated);
|
||||
* - final: SETTLED_MS after its last holder's par2, every holder is final (release_final: a run of done ranges holds
|
||||
* it with hours to spare, so no more of its upload will be read), the candidates of the ranges around it are probed,
|
||||
* and the merger has seen every hash of its files. A file still missing then is one no probe will hash: most sets
|
||||
* describe a readable upload's files, which aren't lone candidates (9 in 10 sets were kept for that, 19 GB).
|
||||
* par2_sets stays (the judge's unusable rule, the merge's holders). A set comes up again only with a new holder: its
|
||||
* par2 reposted, whose probe writes the rows again; without rows its merge finds no files and the count check holds.
|
||||
* The sweep goes through the sets with rows by set_id and starts over at the end: a set kept now (a holder at the edge
|
||||
* of a run, its ranges' probes not done yet) is looked at again on the next pass.
|
||||
*/
|
||||
export class Par2Pruner {
|
||||
/** Sets per batch (BATCH; smaller in the tests). */
|
||||
constructor(private batch = BATCH) {}
|
||||
|
||||
/**
|
||||
* Batches until done or `budgetMs` is spent (the cursor is saved per batch, the next round goes on from it): run
|
||||
* from the merger's loop, a first round over two days of sets held the merges back for hours (LOG.md 2026-10-17 (72)).
|
||||
* Batches until a pass is done or `budgetMs` is spent (the cursor is saved per batch, the next round goes on from
|
||||
* it): run from the merger's loop, a first round over two days of sets held the merges back for hours (LOG.md
|
||||
* 2026-10-17 (72)).
|
||||
*/
|
||||
async round(budgetMs = Infinity): Promise<PruneRound> {
|
||||
const deadline = Date.now() + budgetMs
|
||||
const round: PruneRound = { complete: 0, judged: 0, rows: 0 }
|
||||
// (created_at to the microsecond, set_id): one prober flush gives thousands of sets the same created_at, and a
|
||||
// timestamp read back to the millisecond took the same batch again every round (LOG.md 2026-10-17 (73)).
|
||||
const { rows: cursors } = await db.rawQuery(`SELECT at::text AS at, key FROM worker_cursors WHERE name = ?`, [CURSOR])
|
||||
let since: string = cursors[0]?.at ?? '1970-01-01 00:00:00+00'
|
||||
let sinceKey: string = cursors[0]?.key ?? ''
|
||||
const until = new Date(Date.now() - SETTLED_MS)
|
||||
const round: PruneRound = { complete: 0, judged: 0, final: 0, rows: 0 }
|
||||
const { rows: cursors } = await db.rawQuery(`SELECT key FROM worker_cursors WHERE name = ?`, [CURSOR])
|
||||
let after: string = cursors[0]?.key ?? ''
|
||||
// Ranges done since the oldest merge-kind candidate still unprobed: their releases' hashes and file lists may be
|
||||
// still to come. Hashes since the merger's last window: it hasn't matched them yet.
|
||||
const { rows: marks } = await db.rawQuery(
|
||||
`SELECT coalesce((SELECT min(created_at) FROM probe_candidates WHERE status IN ('waiting', 'taken') AND kind = ANY(?)), now()) AS probed,
|
||||
coalesce((SELECT at FROM worker_cursors WHERE name = ?), to_timestamp(0)) - make_interval(secs => ?) AS merged,
|
||||
now() - make_interval(secs => ?) AS settled`,
|
||||
[MERGE_KINDS, MERGER, MERGER_OVERLAP_MS / 1000, SETTLED_MS / 1000]
|
||||
)
|
||||
const { probed, merged, settled } = marks[0]
|
||||
for (;;) {
|
||||
const { rows: sets } = await db.rawQuery(
|
||||
`SELECT set_id, max(created_at)::text AS at FROM par2_sets GROUP BY set_id
|
||||
HAVING (max(created_at), set_id::text) > (?::timestamptz, ?::text) AND max(created_at) <= ?
|
||||
ORDER BY max(created_at), set_id LIMIT ?`,
|
||||
[since, sinceKey, until, this.batch]
|
||||
`WITH batch AS MATERIALIZED (SELECT DISTINCT set_id FROM par2_files WHERE set_id > ? ORDER BY set_id LIMIT ?),
|
||||
recent AS MATERIALIZED (SELECT group_id, posted_low, posted_high FROM ranges WHERE status = 'done' AND done_at >= ?),
|
||||
unseen AS MATERIALIZED (SELECT DISTINCT hash16k, size FROM file_hashes WHERE created_at >= ?)
|
||||
SELECT b.set_id, CASE
|
||||
WHEN NOT EXISTS (SELECT 1 FROM par2_sets s WHERE s.set_id = b.set_id AND s.created_at > ?)
|
||||
AND NOT EXISTS (SELECT 1 FROM par2_files f WHERE f.set_id = b.set_id AND NOT EXISTS (
|
||||
SELECT 1 FROM file_hashes h JOIN par2_sets s ON s.release_id = h.release_id AND s.set_id = f.set_id
|
||||
WHERE h.hash16k = f.hash16k AND h.size = f.length)) THEN 'complete'
|
||||
WHEN NOT EXISTS (SELECT 1 FROM par2_sets o LEFT JOIN release_verdicts v ON v.release_id = o.release_id AND v.rule = ANY(?)
|
||||
WHERE o.set_id = b.set_id AND v.release_id IS NULL) THEN 'judged'
|
||||
WHEN NOT EXISTS (SELECT 1 FROM par2_sets s WHERE s.set_id = b.set_id AND s.created_at > ?)
|
||||
AND NOT EXISTS (SELECT 1 FROM par2_sets s JOIN releases r ON r.id = s.release_id WHERE s.set_id = b.set_id AND (
|
||||
release_final(r.group_id, r.first_article, r.last_article, r.posted_at, r.last_posted_at, r.complete) IS NOT TRUE
|
||||
OR EXISTS (SELECT 1 FROM recent x WHERE x.group_id = r.group_id
|
||||
AND x.posted_low <= coalesce(r.last_posted_at, r.posted_at) + interval '6 hours' AND x.posted_high >= r.posted_at - interval '6 hours')))
|
||||
AND NOT EXISTS (SELECT 1 FROM par2_files f JOIN unseen u ON u.hash16k = f.hash16k AND u.size = f.length WHERE f.set_id = b.set_id)
|
||||
THEN 'final' END AS reason
|
||||
FROM batch b ORDER BY b.set_id`,
|
||||
[after, this.batch, probed, merged, settled, JUNK, settled]
|
||||
)
|
||||
if (sets.length === 0) break
|
||||
const ids = sets.map((s: { set_id: string }) => s.set_id)
|
||||
const { rows: done } = await db.rawQuery(
|
||||
`SELECT c.set_id FROM unnest(?::char(32)[]) AS c(set_id)
|
||||
WHERE EXISTS (SELECT 1 FROM par2_files f WHERE f.set_id = c.set_id)
|
||||
AND NOT EXISTS (
|
||||
SELECT 1 FROM par2_files f WHERE f.set_id = c.set_id AND NOT EXISTS (
|
||||
SELECT 1 FROM file_hashes h JOIN par2_sets s ON s.release_id = h.release_id AND s.set_id = f.set_id
|
||||
WHERE h.hash16k = f.hash16k AND h.size = f.length))`,
|
||||
[ids]
|
||||
)
|
||||
round.complete += done.length
|
||||
round.rows += await this.drop(done.map((s: { set_id: string }) => s.set_id))
|
||||
since = sets.at(-1).at
|
||||
sinceKey = sets.at(-1).set_id
|
||||
const done = (reason: keyof PruneRound) => sets.filter((s: { reason: string }) => s.reason === reason).map((s: { set_id: string }) => s.set_id)
|
||||
for (const reason of ['complete', 'judged', 'final'] as const) {
|
||||
const ids = done(reason)
|
||||
round[reason] += ids.length
|
||||
round.rows += await this.drop(ids)
|
||||
}
|
||||
// Past the last set: the next batch starts the next pass.
|
||||
after = sets.length < this.batch ? '' : sets.at(-1).set_id
|
||||
await db.rawQuery(
|
||||
`INSERT INTO worker_cursors (name, at, key) VALUES (?, ?::timestamptz, ?) ON CONFLICT (name) DO UPDATE SET at = excluded.at, key = excluded.key`,
|
||||
[CURSOR, since, sinceKey]
|
||||
`INSERT INTO worker_cursors (name, at, key) VALUES (?, now(), ?) ON CONFLICT (name) DO UPDATE SET at = excluded.at, key = excluded.key`,
|
||||
[CURSOR, after]
|
||||
)
|
||||
if (sets.length < this.batch || Date.now() > deadline) break
|
||||
if (after === '' || Date.now() > deadline) break
|
||||
}
|
||||
if (Date.now() > deadline) return round
|
||||
// Judged: no cursor (verdicts come and go); only sets with rows left.
|
||||
const { rows: judged } = await db.rawQuery(
|
||||
`SELECT DISTINCT s.set_id FROM release_verdicts v JOIN par2_sets s ON s.release_id = v.release_id
|
||||
WHERE v.rule = ANY(?)
|
||||
AND EXISTS (SELECT 1 FROM par2_files f WHERE f.set_id = s.set_id)
|
||||
AND NOT EXISTS (
|
||||
SELECT 1 FROM par2_sets o LEFT JOIN release_verdicts ov ON ov.release_id = o.release_id AND ov.rule = ANY(?)
|
||||
WHERE o.set_id = s.set_id AND ov.release_id IS NULL)`,
|
||||
[JUNK, JUNK]
|
||||
)
|
||||
round.judged = judged.length
|
||||
const judgedIds = judged.map((s: { set_id: string }) => s.set_id)
|
||||
for (let i = 0; i < judgedIds.length; i += BATCH) round.rows += await this.drop(judgedIds.slice(i, i + BATCH))
|
||||
return round
|
||||
}
|
||||
|
||||
|
||||
@@ -0,0 +1,17 @@
|
||||
import { BaseSchema } from '@adonisjs/lucid/schema'
|
||||
|
||||
/**
|
||||
* Probe candidates added to a release after it settled (a later range of its upload, a merge into it) stayed not
|
||||
* ready: the settler marks a release's candidates only when it settles it (LOG.md 2026-10-20 (87)). 93,642 waited
|
||||
* since 2026-10-13 (17k lone, 31k par2, 45k archive). The batch store and the merge mark them now; these are made
|
||||
* claimable once.
|
||||
*/
|
||||
export default class extends BaseSchema {
|
||||
async up() {
|
||||
this.schema.raw(`
|
||||
UPDATE probe_candidates pc SET ready = true, unreadable = NOT r.readable FROM releases r
|
||||
WHERE pc.status = 'waiting' AND NOT pc.ready AND r.id = pc.release_id AND r.settled_at IS NOT NULL`)
|
||||
}
|
||||
|
||||
async down() {}
|
||||
}
|
||||
@@ -120,6 +120,18 @@ test.group('BatchStore', (group) => {
|
||||
assert.equal(Number(chunks!.n), 4)
|
||||
})
|
||||
|
||||
test('candidates added to a settled release (a later range of its upload) are claimable at once', async ({ assert }) => {
|
||||
await storeIn(GROUP, 1, upload(10, 'Some.Show.S01E02.1080p.mkv', 2, 'Some.Show.S01E02.1080p [1/2]', '2026-10-05T10:00:00Z'))
|
||||
// Settled before the next range came (the settler marks only what it settles).
|
||||
await db.from('releases').update({ settled_at: new Date() })
|
||||
await db.from('probe_candidates').update({ ready: true })
|
||||
await storeIn(GROUP, 100_001, upload(100_010, 'Some.Show.S01E02.1080p.par2', 1, 'Some.Show.S01E02.1080p [2/2]', '2026-10-05T10:20:00Z'))
|
||||
const candidates = await db.from('probe_candidates').orderBy('id').select('kind', 'ready')
|
||||
assert.deepEqual(candidates.map((c) => c.kind), ['par2'])
|
||||
assert.isTrue(candidates.every((c) => c.ready))
|
||||
assert.lengthOf(await releasesOf(), 1)
|
||||
})
|
||||
|
||||
test('the same upload in another group (multi-posted) is a release of its own', async ({ assert }) => {
|
||||
await storeIn(GROUP, 1, upload(10, 'Some.Show.S01E01.1080p.mkv', 2, 'Some.Show.S01E01.1080p [1/1]', '2026-10-05T10:00:00Z'))
|
||||
await storeIn('alt.binaries.other', 1, upload(10, 'Some.Show.S01E01.1080p.mkv', 2, 'Some.Show.S01E01.1080p [1/1]', '2026-10-05T10:01:00Z', 'alt.binaries.other'))
|
||||
|
||||
@@ -53,23 +53,53 @@ test.group('Par2Pruner', (group) => {
|
||||
const locked = await set('e', e, 2, null)
|
||||
await db.table('release_verdicts').insert({ release_id: e, rule: 'locked', rule_version: 1 })
|
||||
|
||||
// Out of budget: the complete pass's batch, not the judged pass (the next round goes on).
|
||||
assert.deepEqual(await new Par2Pruner().round(0), { complete: 1, judged: 0, rows: 2 })
|
||||
await db.from('worker_cursors').where('name', 'par2-prune').delete()
|
||||
const round = await new Par2Pruner().round()
|
||||
assert.deepEqual(round, { complete: 0, judged: 1, rows: 2 })
|
||||
// Final: its holder held by a run of done ranges with hours to spare, nothing hashed (a readable upload's files).
|
||||
const f = await release()
|
||||
const final = await set('f', f, 2, null)
|
||||
const posted = new Date('2024-03-01T12:00:00Z')
|
||||
const [group] = await db.table('groups').insert({ name: 'alt.binaries.pruned', priority: 1 }).returning('id')
|
||||
const groupId = Number(group.id)
|
||||
await db.from('releases').where('id', f).update({ group_id: groupId, posted_at: posted, last_posted_at: posted, first_article: 5000, last_article: 5100 })
|
||||
const [range] = await db
|
||||
.table('ranges')
|
||||
.insert({ group_id: groupId, start_id: 1, end_id: 10000, next_id: 10001, status: 'done', done_at: old, posted_low: new Date('2024-02-28T00:00:00Z'), posted_high: new Date('2024-03-03T00:00:00Z') })
|
||||
.returning('id')
|
||||
await db.table('range_blocks').insert({ group_id: groupId, start_id: 1, end_id: 10000, posted_low: new Date('2024-02-28T00:00:00Z'), posted_high: new Date('2024-03-03T00:00:00Z'), at_group_start: false })
|
||||
// A lone candidate unprobed since before the range was done: its hash may still match. Kept.
|
||||
await db.table('probe_candidates').insert({ id: 9001, release_id: b, kind: 'lone', message_id: '<9001@x>', file: 'x', status: 'waiting', created_at: new Date(old.getTime() - 3600_000) })
|
||||
|
||||
// Out of budget: one batch of two sets (a, b), the next round goes on.
|
||||
assert.deepEqual(await new Par2Pruner(2).round(0), { complete: 1, judged: 0, final: 0, rows: 2 })
|
||||
assert.deepEqual(await new Par2Pruner(2).round(), { complete: 0, judged: 1, final: 0, rows: 2 })
|
||||
assert.equal(await count(complete), 0)
|
||||
assert.equal(await count(apart), 2)
|
||||
assert.equal(await count(recent), 2)
|
||||
assert.equal(await count(dead), 0)
|
||||
assert.equal(await count(locked), 2)
|
||||
assert.equal((await db.from('par2_sets').count('* as n'))[0].n, '5')
|
||||
assert.equal(await count(final), 2)
|
||||
|
||||
// Again: the cursor is past them, nothing more.
|
||||
assert.deepEqual(await new Par2Pruner().round(), { complete: 0, judged: 0, rows: 0 })
|
||||
// Probed: final. The next pass starts over from the first set.
|
||||
await db.from('probe_candidates').where('id', 9001).update({ status: 'done' })
|
||||
assert.deepEqual(await new Par2Pruner(2).round(), { complete: 0, judged: 0, final: 1, rows: 2 })
|
||||
assert.equal(await count(final), 0)
|
||||
assert.equal((await db.from('par2_sets').count('* as n'))[0].n, '6')
|
||||
|
||||
// A hash of its files since the merger's last window: the merger hasn't matched it yet. Kept.
|
||||
const g = await release()
|
||||
const unseen = await set('g', g, 1, null)
|
||||
await db.from('releases').where('id', g).update({ group_id: groupId, posted_at: posted, last_posted_at: posted, first_article: 6000, last_article: 6100 })
|
||||
await db.table('worker_cursors').insert({ name: 'merger', at: new Date(Date.now() - 3600_000) })
|
||||
await db.table('probe_candidates').insert({ id: 9002, release_id: await release(), kind: 'lone', message_id: '<9002@x>', file: 'x', status: 'done' })
|
||||
await db.table('file_hashes').insert({ candidate_id: 9002, release_id: 1, hash16k: 'g0'.padEnd(32, '0'), size: 1000 })
|
||||
assert.equal((await new Par2Pruner().round()).final, 0)
|
||||
// Past the hash by more than the merger's overlap.
|
||||
await db.from('worker_cursors').where('name', 'merger').update({ at: new Date(Date.now() + 120_000) })
|
||||
assert.equal((await new Par2Pruner().round()).final, 1)
|
||||
assert.equal(await count(unseen), 0)
|
||||
assert.isNotNull(range)
|
||||
})
|
||||
|
||||
test('more sets of one created_at than a batch holds: each batch goes on past the last one', async ({ assert }) => {
|
||||
test('sets of one created_at, a batch each: every one looked at', async ({ assert }) => {
|
||||
// One prober flush gives thousands of sets the same created_at (to the microsecond).
|
||||
const flush = new Date('2026-10-15T14:38:39.883Z')
|
||||
let candidate = 7000
|
||||
@@ -88,6 +118,7 @@ test.group('Par2Pruner', (group) => {
|
||||
const round = await new Par2Pruner(1).round()
|
||||
assert.equal(round.complete, 3)
|
||||
assert.equal((await db.from('par2_files').whereIn('set_id', ['p'.repeat(32), 'q'.repeat(32), 'r'.repeat(32)]).count('* as n'))[0].n, '0')
|
||||
assert.equal((await db.from('worker_cursors').where('name', 'par2-prune').firstOrFail()).key, 'r'.repeat(32))
|
||||
// The pass ended: the next one starts from the first set.
|
||||
assert.equal((await db.from('worker_cursors').where('name', 'par2-prune').firstOrFail()).key, '')
|
||||
})
|
||||
})
|
||||
Reference in new issue
Block a user