The merger keeps its place over a restart (worker_cursors), not from 1970 each deploy

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

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<MergeRound> {
     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
   }
 

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`)
+  }
+}

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[] = []