diff --git a/app/services/par2_pruner.ts b/app/services/par2_pruner.ts index f8af8e7..5e4423c 100644 --- a/app/services/par2_pruner.ts +++ b/app/services/par2_pruner.ts @@ -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. */ 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)). @@ -33,14 +36,18 @@ export class Par2Pruner { async round(budgetMs = Infinity): Promise { const deadline = Date.now() + budgetMs const round: PruneRound = { complete: 0, judged: 0, rows: 0 } - const saved = await db.from('worker_cursors').where('name', CURSOR).first() - let since: Date = saved ? new Date(saved.at) : new Date(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) for (;;) { const { rows: sets } = await db.rawQuery( - `SELECT set_id, max(created_at) AS at FROM par2_sets GROUP BY set_id - HAVING max(created_at) > ? AND max(created_at) <= ? ORDER BY 2 LIMIT ?`, - [since, until, BATCH] + `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] ) if (sets.length === 0) break const ids = sets.map((s: { set_id: string }) => s.set_id) @@ -56,11 +63,12 @@ export class Par2Pruner { 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 await db.rawQuery( - `INSERT INTO worker_cursors (name, at) VALUES (?, ?) ON CONFLICT (name) DO UPDATE SET at = excluded.at`, - [CURSOR, since] + `INSERT INTO worker_cursors (name, at, key) VALUES (?, ?::timestamptz, ?) ON CONFLICT (name) DO UPDATE SET at = excluded.at, key = excluded.key`, + [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 // Judged: no cursor (verdicts come and go); only sets with rows left. diff --git a/database/migrations/1770000004300_worker_cursors_key.ts b/database/migrations/1770000004300_worker_cursors_key.ts new file mode 100644 index 0000000..b00122a --- /dev/null +++ b/database/migrations/1770000004300_worker_cursors_key.ts @@ -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`) + } +} diff --git a/tests/functional/par2_pruner.spec.ts b/tests/functional/par2_pruner.spec.ts index f19feb1..8cafeae 100644 --- a/tests/functional/par2_pruner.spec.ts +++ b/tests/functional/par2_pruner.spec.ts @@ -68,4 +68,26 @@ test.group('Par2Pruner', (group) => { // Again: the cursor is past them, nothing more. 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)) + }) })