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 <noreply@anthropic.com>
This commit is contained in:
1 parent
c55918350e
commit
aa20d6b79f
6 files changed
+41
-4
No files matched your search
@@ -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<number> {
|
||||
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`)
|
||||
|
||||
@@ -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()
|
||||
|
||||
@@ -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<PruneRound> {
|
||||
/**
|
||||
* 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<PruneRound> {
|
||||
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
|
||||
|
||||
@@ -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}`)
|
||||
|
||||
@@ -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')
|
||||
})
|
||||
})
|
||||
@@ -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)
|
||||
|
||||
Reference in new issue
Block a user