The par2 prune's cursor is (created_at to the microsecond, set_id): 20k sets of one flush share a created_at
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
1 parent
f2c5607ec3
commit
a5e0399ee6
3 files changed
+54
-8
No files matched your search
@@ -26,6 +26,9 @@ export interface PruneRound {
|
|||||||
* par2 reposted, whose probe writes the rows again; without rows its merge finds no files and the count check holds.
|
* par2 reposted, whose probe writes the rows again; without rows its merge finds no files and the count check holds.
|
||||||
*/
|
*/
|
||||||
export class Par2Pruner {
|
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
|
* 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)).
|
* 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)).
|
||||||
@@ -33,14 +36,18 @@ export class Par2Pruner {
|
|||||||
async round(budgetMs = Infinity): Promise<PruneRound> {
|
async round(budgetMs = Infinity): Promise<PruneRound> {
|
||||||
const deadline = Date.now() + budgetMs
|
const deadline = Date.now() + budgetMs
|
||||||
const round: PruneRound = { complete: 0, judged: 0, rows: 0 }
|
const round: PruneRound = { complete: 0, judged: 0, rows: 0 }
|
||||||
const saved = await db.from('worker_cursors').where('name', CURSOR).first()
|
// (created_at to the microsecond, set_id): one prober flush gives thousands of sets the same created_at, and a
|
||||||
let since: Date = saved ? new Date(saved.at) : new Date(0)
|
// 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 until = new Date(Date.now() - SETTLED_MS)
|
||||||
for (;;) {
|
for (;;) {
|
||||||
const { rows: sets } = await db.rawQuery(
|
const { rows: sets } = await db.rawQuery(
|
||||||
`SELECT set_id, max(created_at) AS at FROM par2_sets GROUP BY set_id
|
`SELECT set_id, max(created_at)::text AS at FROM par2_sets GROUP BY set_id
|
||||||
HAVING max(created_at) > ? AND max(created_at) <= ? ORDER BY 2 LIMIT ?`,
|
HAVING (max(created_at), set_id::text) > (?::timestamptz, ?::text) AND max(created_at) <= ?
|
||||||
[since, until, BATCH]
|
ORDER BY max(created_at), set_id LIMIT ?`,
|
||||||
|
[since, sinceKey, until, this.batch]
|
||||||
)
|
)
|
||||||
if (sets.length === 0) break
|
if (sets.length === 0) break
|
||||||
const ids = sets.map((s: { set_id: string }) => s.set_id)
|
const ids = sets.map((s: { set_id: string }) => s.set_id)
|
||||||
@@ -56,11 +63,12 @@ export class Par2Pruner {
|
|||||||
round.complete += done.length
|
round.complete += done.length
|
||||||
round.rows += await this.drop(done.map((s: { set_id: string }) => s.set_id))
|
round.rows += await this.drop(done.map((s: { set_id: string }) => s.set_id))
|
||||||
since = sets.at(-1).at
|
since = sets.at(-1).at
|
||||||
|
sinceKey = sets.at(-1).set_id
|
||||||
await db.rawQuery(
|
await db.rawQuery(
|
||||||
`INSERT INTO worker_cursors (name, at) VALUES (?, ?) ON CONFLICT (name) DO UPDATE SET at = excluded.at`,
|
`INSERT INTO worker_cursors (name, at, key) VALUES (?, ?::timestamptz, ?) ON CONFLICT (name) DO UPDATE SET at = excluded.at, key = excluded.key`,
|
||||||
[CURSOR, since]
|
[CURSOR, since, sinceKey]
|
||||||
)
|
)
|
||||||
if (sets.length < BATCH || Date.now() > deadline) break
|
if (sets.length < this.batch || Date.now() > deadline) break
|
||||||
}
|
}
|
||||||
if (Date.now() > deadline) return round
|
if (Date.now() > deadline) return round
|
||||||
// Judged: no cursor (verdicts come and go); only sets with rows left.
|
// Judged: no cursor (verdicts come and go); only sets with rows left.
|
||||||
|
|||||||
@@ -0,0 +1,16 @@
|
|||||||
|
import { BaseSchema } from '@adonisjs/lucid/schema'
|
||||||
|
|
||||||
|
/**
|
||||||
|
* worker_cursors.key: a tie-breaker after `at`. The par2 pruner's cursor was a timestamp alone, read back to the
|
||||||
|
* millisecond: 20,000 sets of one prober flush share created_at 2026-10-15 14:38:39.883984, and every round took the
|
||||||
|
* same batch again (LOG.md 2026-10-17 (73)). Now (at to the microsecond, set_id).
|
||||||
|
*/
|
||||||
|
export default class extends BaseSchema {
|
||||||
|
async up() {
|
||||||
|
this.schema.raw(`ALTER TABLE worker_cursors ADD COLUMN IF NOT EXISTS key text`)
|
||||||
|
}
|
||||||
|
|
||||||
|
async down() {
|
||||||
|
this.schema.raw(`ALTER TABLE worker_cursors DROP COLUMN IF EXISTS key`)
|
||||||
|
}
|
||||||
|
}
|
||||||
@@ -68,4 +68,26 @@ test.group('Par2Pruner', (group) => {
|
|||||||
// Again: the cursor is past them, nothing more.
|
// Again: the cursor is past them, nothing more.
|
||||||
assert.deepEqual(await new Par2Pruner().round(), { complete: 0, judged: 0, rows: 0 })
|
assert.deepEqual(await new Par2Pruner().round(), { complete: 0, judged: 0, rows: 0 })
|
||||||
})
|
})
|
||||||
|
|
||||||
|
test('more sets of one created_at than a batch holds: each batch goes on past the last one', 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
|
||||||
|
for (const letter of ['p', 'q', 'r']) {
|
||||||
|
const key = `pr${letter}`.padEnd(40, '0')
|
||||||
|
const [row] = await db.table('releases').insert({ release_key: key, identity_key: key, name: key, source: 'stem', groups: '[]', group_id: 1, files: 1, readable: false }).returning('id')
|
||||||
|
const id = Number(row.id)
|
||||||
|
const setId = letter.repeat(32)
|
||||||
|
const h = `${letter}0`.padEnd(32, '0')
|
||||||
|
await db.table('par2_files').insert({ set_id: setId, file_id: h, hash16k: h, length: 1000, name: 'f.rar' })
|
||||||
|
candidate++
|
||||||
|
await db.table('probe_candidates').insert({ id: candidate, release_id: id, kind: 'lone', message_id: `<${candidate}@x>`, file: 'x', status: 'done' })
|
||||||
|
await db.table('file_hashes').insert({ candidate_id: candidate, release_id: id, hash16k: h, size: 1000 })
|
||||||
|
await db.rawQuery(`INSERT INTO par2_sets (set_id, release_id, files, created_at) VALUES (?, ?, 1, ?::timestamptz + interval '984 microseconds')`, [setId, id, flush])
|
||||||
|
}
|
||||||
|
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))
|
||||||
|
})
|
||||||
})
|
})
|
||||||
Reference in new issue
Block a user