diff --git a/app/services/par2_merger.ts b/app/services/par2_merger.ts index 2b9cde0..7f23a68 100644 --- a/app/services/par2_merger.ts +++ b/app/services/par2_merger.ts @@ -18,6 +18,8 @@ const SLACK_FILES = 10 const ROUND_MS = 60_000 /** A round looks back this far past the previous one's start (rows committed while it ran). */ const OVERLAP_MS = 60_000 +/** The merger's row in worker_cursors: the start of its last whole round, kept over a restart. */ +const CURSOR = 'merger' export interface MergeRound { sets: number @@ -73,7 +75,8 @@ interface Member { export class Par2Merger { private stopped = false private folder = new SidecarFolder((trx, found, key, via) => this.merge(trx, found, key, via)) - private since = new Date(0) + /** Null until read from worker_cursors (1970 the first time: everything). */ + private since: Date | null = null stop() { this.stopped = true @@ -97,6 +100,10 @@ export class Par2Merger { /** The sets with something new since the last round: a par2 found, or a hash of one of their files. */ async round(): Promise { const { rows: now } = await db.rawQuery(`SELECT now() AS at`) + if (!this.since) { + const saved = await db.from('worker_cursors').where('name', CURSOR).first() + this.since = saved ? new Date(saved.at) : new Date(0) + } const since = new Date(this.since.getTime() - OVERLAP_MS) const { rows } = await db.rawQuery( `SELECT set_id FROM par2_sets WHERE created_at >= ? @@ -136,7 +143,12 @@ export class Par2Merger { } // Third: sidecars (.nfo, .sfv, .nzb) apart from their upload join it. if (!this.stopped) round.folded = await this.folder.round() + // Only after a whole round: a restart halfway goes on from the round before. this.since = now[0].at + await db.rawQuery( + `INSERT INTO worker_cursors (name, at) VALUES (?, ?) ON CONFLICT (name) DO UPDATE SET at = excluded.at`, + [CURSOR, this.since] + ) return round } diff --git a/database/migrations/1770000003600_worker_cursors.ts b/database/migrations/1770000003600_worker_cursors.ts new file mode 100644 index 0000000..bb5e3a8 --- /dev/null +++ b/database/migrations/1770000003600_worker_cursors.ts @@ -0,0 +1,20 @@ +import { BaseSchema } from '@adonisjs/lucid/schema' + +/** + * Where a worker's incremental rounds go on after a restart. The merger kept its "since" in memory only: each deploy + * or pause started it from 1970, a first round over every par2 set and inner stem ever found (25+ min with par2_files + * on the HDD) before the sidecar fold's turn (LOG.md 2026-10-15 (58)). + */ +export default class extends BaseSchema { + async up() { + this.schema.raw(` + CREATE TABLE worker_cursors ( + name text PRIMARY KEY, + at timestamptz NOT NULL + )`) + } + + async down() { + this.schema.raw(`DROP TABLE IF EXISTS worker_cursors`) + } +} diff --git a/tests/functional/par2_merger.spec.ts b/tests/functional/par2_merger.spec.ts index 4c40086..773b17e 100644 --- a/tests/functional/par2_merger.spec.ts +++ b/tests/functional/par2_merger.spec.ts @@ -65,6 +65,19 @@ test.group('Par2Merger', (group) => { assert.deepEqual(await new Par2Merger().round().then((r) => r.merges), 0) }) + test('a restart goes on from the last whole round, kept in worker_cursors', async ({ assert }) => { + const hourAgo = new Date(Date.now() - 3600_000) + await db.table('par2_sets').insert({ set_id: 'f'.repeat(32), release_id: 1, files: 2, created_at: hourAgo }) + // Never run: everything. + assert.equal((await new Par2Merger().round()).sets, 1) + const saved = await db.from('worker_cursors').where('name', 'merger').firstOrFail() + assert.isAbove(new Date(saved.at).getTime(), hourAgo.getTime()) + // A new process (a deploy) reads where the last round started: the set an hour old is behind it. + assert.equal((await new Par2Merger().round()).sets, 0) + await db.from('worker_cursors').where('name', 'merger').update({ at: new Date(hourAgo.getTime() - 60_000) }) + assert.equal((await new Par2Merger().round()).sets, 1) + }) + test('a set held by two of the releases merged is held once by the target', async ({ assert }) => { const at = new Date('2022-07-08T03:46:00Z') const ids: number[] = []