From aa20d6b79f54bdc1e1ee7c82b73c4fe7433232bb Mon Sep 17 00:00:00 2001 From: Daan Meijer Date: Sat, 17 Oct 2026 20:42:14 +0200 Subject: [PATCH] The par2 prune gets 2 min per round (it held the merges back for hours); abandoned ranges past their end are finished Co-Authored-By: Claude Opus 5.5 --- app/services/leases.ts | 16 ++++++++++++++++ app/services/par2_merger.ts | 4 +++- app/services/par2_pruner.ts | 10 ++++++++-- commands/coordinator_run.ts | 3 +++ tests/functional/leases.spec.ts | 7 +++++++ tests/functional/par2_pruner.spec.ts | 5 ++++- 6 files changed, 41 insertions(+), 4 deletions(-) diff --git a/app/services/leases.ts b/app/services/leases.ts index 5c34f2a..8afc17e 100644 --- a/app/services/leases.ts +++ b/app/services/leases.ts @@ -52,6 +52,22 @@ class Leases { return { range: { id: Number(row.id), group: group.name, start: Number(row.start_id), end: Number(row.end_id), next: Number(row.next_id) }, rules } } + /** + * Ranges read to their end whose worker went quiet before finishing them: done. The lease query skips a range past + * its end, so one whose finish never came stayed leased for good (a gap range since 2026-10-16 14:03, during the lock + * pile-up; LOG.md 2026-10-17 (72)). Its batches are all stored; its spooled chunks go. + */ + async finishAbandoned(): Promise { + const { rows } = await db.rawQuery( + `UPDATE ranges SET status = 'done', done_at = now() + WHERE status = 'leased' AND next_id > end_id AND heartbeat_at < now() - make_interval(secs => ?) + RETURNING id`, + [coordinatorConfig.leaseTtlSeconds] + ) + for (const { id } of rows) await Spool.remove(Number(id)) + return rows.length + } + async leased(rangeId: number, nodeId: number) { const range = await db.from('ranges').where('id', rangeId).first() if (!range || range.status !== 'leased' || Number(range.node_id) !== nodeId) throw new LeaseLost(`Range ${rangeId} is not leased to this node`) diff --git a/app/services/par2_merger.ts b/app/services/par2_merger.ts index d0c899d..99ad50d 100644 --- a/app/services/par2_merger.ts +++ b/app/services/par2_merger.ts @@ -23,6 +23,8 @@ const OVERLAP_MS = 60_000 const FOLD_EVERY_MS = 5 * 60_000 /** Between par2_files prunes (complete and judged sets' file lists). */ const PRUNE_EVERY_MS = 60 * 60_000 +/** A prune's share of the merger's time: the rest goes on in the next one. */ +const PRUNE_BUDGET_MS = 2 * 60_000 /** The merger's row in worker_cursors: the start of its last whole round, kept over a restart. */ const CURSOR = 'merger' @@ -95,7 +97,7 @@ export class Par2Merger { // First, at start too: a prune needn't wait for a long round. if (!this.stopped && Date.now() - prunedAt >= PRUNE_EVERY_MS) { prunedAt = Date.now() - const pruned = await new Par2Pruner().round() + 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.`) } const round = await this.round() diff --git a/app/services/par2_pruner.ts b/app/services/par2_pruner.ts index b12b6c5..f8af8e7 100644 --- a/app/services/par2_pruner.ts +++ b/app/services/par2_pruner.ts @@ -26,7 +26,12 @@ 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 { - async round(): Promise { + /** + * 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)). + */ + 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) @@ -55,8 +60,9 @@ export class Par2Pruner { `INSERT INTO worker_cursors (name, at) VALUES (?, ?) ON CONFLICT (name) DO UPDATE SET at = excluded.at`, [CURSOR, since] ) - if (sets.length < BATCH) break + if (sets.length < BATCH || 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 diff --git a/commands/coordinator_run.ts b/commands/coordinator_run.ts index 448bb51..55e2f4f 100644 --- a/commands/coordinator_run.ts +++ b/commands/coordinator_run.ts @@ -6,6 +6,7 @@ import Planner from '#services/planner' import FragmentMerger from '#services/fragment_merger' import Settler from '#services/settler' import RangeBlocks from '#services/range_blocks' +import Leases from '#services/leases' import NntpPool from '#services/nntp_pool' import coordinatorConfig from '#config/coordinator' @@ -70,6 +71,8 @@ export default class CoordinatorRun extends BaseCommand { const blocks = (async () => { while (main && !stopping) { try { + const abandoned = await Leases.finishAbandoned() + if (abandoned > 0) this.logger.info(`Finished ${abandoned} ranges read to their end whose worker went quiet.`) await RangeBlocks.rebuild() } catch (error: any) { this.logger.error(`Range blocks failed: ${error.message}`) diff --git a/tests/functional/leases.spec.ts b/tests/functional/leases.spec.ts index 6779711..ec0e513 100644 --- a/tests/functional/leases.spec.ts +++ b/tests/functional/leases.spec.ts @@ -31,5 +31,12 @@ test.group('Leases', (group) => { // a1's worker went quiet: its lease is taken over (a2 alone counts). await db.from('ranges').where('id', a1).update({ heartbeat_at: new Date(Date.now() - 24 * 3600_000) }) assert.equal(await leased(5), a1) + + // Read to its end, its worker gone before finishing: done. One still reading, or quiet but not at its end, stays. + await db.from('ranges').where('id', a2).update({ next_id: 201, heartbeat_at: new Date(Date.now() - 24 * 3600_000) }) + await db.from('ranges').where('id', b1).update({ next_id: 101 }) + assert.equal(await Leases.finishAbandoned(), 1) + assert.equal((await db.from('ranges').where('id', a2).firstOrFail()).status, 'done') + assert.equal((await db.from('ranges').where('id', b1).firstOrFail()).status, 'leased') }) }) diff --git a/tests/functional/par2_pruner.spec.ts b/tests/functional/par2_pruner.spec.ts index 3d91ebb..f19feb1 100644 --- a/tests/functional/par2_pruner.spec.ts +++ b/tests/functional/par2_pruner.spec.ts @@ -53,8 +53,11 @@ 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: 1, judged: 1, rows: 4 }) + assert.deepEqual(round, { complete: 0, judged: 1, rows: 2 }) assert.equal(await count(complete), 0) assert.equal(await count(apart), 2) assert.equal(await count(recent), 2)