Merger: what's new an hour at a time, only the sets two releases share; common files match nothing

A round behind by two days of the probe backlog read 3.5M sets and never ended;
a par2 file shared by up to 190k files loaded every release it matched and the
merger ran out of heap every few minutes (LOG.md 2026-10-20 (85)).

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

REBUILD.md

@@ -238,6 +238,10 @@ files the set describes, into the one with the most files (chunks,
 candidates, inspections, counts; the par2's shared name when none was
 readable); `release_merges` keeps what went where. Obfuscated uploads whose
 files were each a release of their own (`lone` candidates) become one.
+It reads what's new an hour (`created_at`) at a time, its place kept after
+each, and merges only the sets held or matched by two releases or more; a
+file whose first 16 KiB and size more than 20 files share is a common one and
+matches nothing (LOG.md 2026-10-20 (85)).
 
 **Backlog: voyager 1's features not in voyager 2** (CONCEPT.md decisions;
 voyager 1's data exported to `/mnt/bulk/usenet-rebuild/v1-export/` on

app/services/par2_merger.ts

@@ -15,17 +15,28 @@ const MAX_HOLDERS = 4
 /** Members with more files than this many times the set's, plus SLACK_FILES (its par2 volumes), are more than one upload. */
 const MAX_FILES_PER_SET_FILE = 3
 const SLACK_FILES = 10
+/**
+ * A par2 file held by more file rows than this (the same first 16 KiB and size) is a common file, not one upload's:
+ * it matches nothing. One such file was in 190,380 uploads, and a set naming it loaded every one of them; the merger
+ * ran out of heap every few minutes (LOG.md 2026-10-20 (85)). The judge calls 20 releases repeated.
+ */
+const COMMON_FILE = 20
 /** Between rounds. */
 const ROUND_MS = 60_000
-/** A round looks back this far past the previous one's start (rows committed while it ran). */
+/** A window looks back this far past the previous one's end (rows committed after it was read). */
 const OVERLAP_MS = 60_000
+/**
+ * A round goes through what's new in windows of this much (created_at), its place kept after each: a round behind by
+ * two days of the probe backlog (3.5M sets) never ended, and every restart started it over (LOG.md 2026-10-20 (85)).
+ */
+const WINDOW_MS = 60 * 60_000
 /** The sidecar fold also runs this often inside a long par2 pass: its queue grows by the settler's pace. */
 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. */
+/** The merger's row in worker_cursors: the end of its last whole window, kept over a restart. */
 const CURSOR = 'merger'
 
 export interface MergeRound {
@@ -46,6 +57,12 @@ export interface MergeRound {
  */
 const INNER_STEM = `substring(i.result->>'yencName' from '^([A-Za-z0-9]{16,64})\\.')`
 
+/** The releases holding a par2 file f (its first 16 KiB and size), none when it's a common file (COMMON_FILE). */
+const HOLDERS = `SELECT c.release_id FROM (
+    SELECT l.release_id, count(*) OVER () AS holders
+    FROM (SELECT h.release_id FROM file_hashes h WHERE h.hash16k = f.hash16k AND h.size = f.length LIMIT ${COMMON_FILE + 1}) l
+  ) c WHERE c.holders <= ${COMMON_FILE}`
+
 interface Member {
   id: number
   release_key: string
@@ -111,14 +128,18 @@ export class Par2Merger {
     }
   }
 
-  /** The sets with something new since the last round: a par2 found, or a hash of one of their files. */
+  /**
+   * The sets and inner stems with something new since the last round (a par2 found, a hash of one of their files, a
+   * lone inspection), a window at a time.
+   */
   async round(): Promise<MergeRound> {
-    const { rows: now } = await db.rawQuery(`SELECT now() AS at`)
+    // A Date keeps milliseconds, now() microseconds: rounded up, a row made at now() is before the end.
+    const { rows: now } = await db.rawQuery(`SELECT now() + interval '1 millisecond' AS at`)
+    const end = new Date(now[0].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)
     // First: sidecars (.nfo, .sfv, .nzb) apart from their upload join it. A queue, not since: it needn't wait for the
     // par2 pass (hours when that starts over).
     const folded = await this.folder.round()
@@ -132,52 +153,96 @@ export class Par2Merger {
       foldedAt = Date.now()
       if (more.folded) logger.info(`Folded ${more.folded} sidecars into their uploads (${more.sidecars} sidecar-only of ${more.looked} small releases).`)
     }
-    const { rows } = await db.rawQuery(
-      `SELECT set_id FROM par2_sets WHERE created_at >= ?
-       UNION
-       SELECT f.set_id FROM file_hashes h JOIN par2_files f ON f.hash16k = h.hash16k AND f.length = h.size WHERE h.created_at >= ?`,
-      [since, since]
-    )
-    const round: MergeRound = { sets: rows.length, merges: 0, releases: 0, stems: 0, stemMerges: 0, stemReleases: 0, folded }
-    for (const { set_id: setId } of rows) {
+    const round: MergeRound = { sets: 0, merges: 0, releases: 0, stems: 0, stemMerges: 0, stemReleases: 0, folded }
+    while (this.since < end) {
       if (this.stopped) return round
-      await foldAgain()
-      // One set that fails doesn't hold up the others (nor the next round, which would start from the same point).
-      const merged = await this.mergeSet(setId.trim()).catch((error) => {
-        logger.error(`Par2 merge of set ${setId.trim()} failed: ${error.message}`)
-        return 0
-      })
-      if (merged > 0) {
-        round.merges++
-        round.releases += merged + 1
+      const from = await this.firstNew(new Date(this.since.getTime() - OVERLAP_MS), end)
+      const until = from ? new Date(Math.min(from.getTime() + WINDOW_MS, end.getTime())) : end
+      if (from) {
+        const { touched, sets } = await this.setsToMerge(from, until)
+        round.sets += touched
+        for (const setId of sets) {
+          if (this.stopped) return round
+          await foldAgain()
+          // One set that fails doesn't hold up the others (nor the next round, which would start from the same point).
+          const merged = await this.mergeSet(setId.trim()).catch((error) => {
+            logger.error(`Par2 merge of set ${setId.trim()} failed: ${error.message}`)
+            return 0
+          })
+          if (merged > 0) {
+            round.merges++
+            round.releases += merged + 1
+          }
+        }
+        const { rows: stems } = await db.rawQuery(
+          `SELECT DISTINCT ${INNER_STEM} AS stem FROM release_inspections i
+           WHERE i.kind = 'lone' AND i.created_at >= ? AND i.created_at <= ? AND ${INNER_STEM} IS NOT NULL`,
+          [from, until]
+        )
+        for (const { stem } of stems) {
+          if (this.stopped) return round
+          if (!looksRandom(stem)) continue
+          round.stems++
+          const merged = await this.mergeStem(stem).catch((error) => {
+            logger.error(`Inner-name merge of ${stem} failed: ${error.message}`)
+            return 0
+          })
+          if (merged > 0) {
+            round.stemMerges++
+            round.stemReleases += merged + 1
+          }
+        }
       }
+      // Only after a whole window: a restart halfway goes on from the window before.
+      this.since = until
+      await db.rawQuery(
+        `INSERT INTO worker_cursors (name, at) VALUES (?, ?) ON CONFLICT (name) DO UPDATE SET at = excluded.at`,
+        [CURSOR, this.since]
+      )
     }
-    const { rows: stems } = await db.rawQuery(
-      `SELECT DISTINCT ${INNER_STEM} AS stem FROM release_inspections i WHERE i.kind = 'lone' AND i.created_at >= ? AND ${INNER_STEM} IS NOT NULL`,
-      [since]
-    )
-    for (const { stem } of stems) {
-      if (this.stopped) return round
-      if (!looksRandom(stem)) continue
-      round.stems++
-      const merged = await this.mergeStem(stem).catch((error) => {
-        logger.error(`Inner-name merge of ${stem} failed: ${error.message}`)
-        return 0
-      })
-      if (merged > 0) {
-        round.stemMerges++
-        round.stemReleases += merged + 1
-      }
-    }
-    // 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
   }
 
+  /** When the first row a window reads was made, from `from` up to `end`; null when none (a gap, the nightly pause). */
+  private async firstNew(from: Date, end: Date): Promise<Date | null> {
+    const { rows } = await db.rawQuery(
+      `SELECT least(
+         (SELECT min(created_at) FROM par2_sets WHERE created_at >= ?),
+         (SELECT min(created_at) FROM file_hashes WHERE created_at >= ?),
+         (SELECT min(created_at) FROM release_inspections WHERE created_at >= ?)) AS at`,
+      [from, from, from]
+    )
+    const at = rows[0].at ? new Date(rows[0].at) : null
+    return at && at <= end ? at : null
+  }
+
+  /**
+   * The sets a window touched (a par2 found, a hash of one of their files), and those among them a merge could
+   * change: held or matched by two releases or more. Most have every file in one release already (96% of a backlog
+   * hour's 147k): mergeSet needn't look at them.
+   */
+  private async setsToMerge(from: Date, until: Date): Promise<{ touched: number; sets: string[] }> {
+    const { rows } = await db.rawQuery(
+      `WITH touched AS MATERIALIZED (
+         SELECT set_id FROM par2_sets WHERE created_at >= ? AND created_at <= ?
+         UNION
+         SELECT f.set_id FROM file_hashes h JOIN par2_files f ON f.hash16k = h.hash16k AND f.length = h.size
+         WHERE h.created_at >= ? AND h.created_at <= ?
+       )
+       SELECT (SELECT count(*) FROM touched) AS touched, array(
+         SELECT t.set_id FROM touched t
+         WHERE EXISTS (
+           SELECT 1 FROM (
+             SELECT s.release_id FROM par2_sets s WHERE s.set_id = t.set_id
+             UNION ALL
+             SELECT m.release_id FROM par2_files f CROSS JOIN LATERAL (${HOLDERS}) m WHERE f.set_id = t.set_id
+           ) u HAVING count(DISTINCT u.release_id) >= 2)
+       ) AS sets`,
+      [from, until, from, until]
+    )
+    return { touched: Number(rows[0].touched), sets: rows[0].sets }
+  }
+
   /** Merges the releases whose lone files' inner names share a random stem; how many went into another. */
   async mergeStem(stem: string): Promise<number> {
     return db.transaction(async (trx) => {
@@ -222,7 +287,7 @@ export class Par2Merger {
       const near = (r: Member) => r.group_id === anchor.group_id && (at(r) === null || at(anchor) === null || Math.abs(at(r)! - at(anchor)!) <= MAX_SPREAD_MS)
       const { rows: matches } = await trx.rawQuery(
         `SELECT f.file_id, r.id, r.release_key, r.group_id, r.files, r.posted_at, r.readable
-         FROM par2_files f JOIN file_hashes h ON h.hash16k = f.hash16k AND h.size = f.length JOIN releases r ON r.id = h.release_id
+         FROM par2_files f CROSS JOIN LATERAL (${HOLDERS}) h JOIN releases r ON r.id = h.release_id
          WHERE f.set_id = ?`,
         [setId]
       )

tests/functional/par2_merger.spec.ts

@@ -78,6 +78,44 @@ test.group('Par2Merger', (group) => {
     assert.equal((await new Par2Merger().round()).sets, 1)
   })
 
+  test('a round behind by hours goes through them a window at a time, past the gaps', async ({ assert }) => {
+    const ago = (hours: number) => new Date(Date.now() - hours * 3600_000)
+    await db.table('par2_sets').multiInsert([
+      { set_id: 'a1'.repeat(16), release_id: 1, files: 1, created_at: ago(30) },
+      { set_id: 'a2'.repeat(16), release_id: 2, files: 1, created_at: ago(2.5) },
+      { set_id: 'a3'.repeat(16), release_id: 3, files: 1, created_at: ago(0.5) },
+    ])
+    await db.table('worker_cursors').insert({ name: 'merger', at: ago(48) })
+    assert.equal((await new Par2Merger().round()).sets, 3)
+    const saved = await db.from('worker_cursors').where('name', 'merger').firstOrFail()
+    assert.isAbove(new Date(saved.at).getTime(), ago(0.5).getTime())
+  })
+
+  test('a par2 file held by more than 20 files is a common one: it merges nothing', async ({ assert }) => {
+    const at = new Date('2022-07-08T03:46:00Z')
+    const ids: number[] = []
+    for (const key of ['dd000001', 'dd000002']) {
+      const [row] = await db
+        .table('releases')
+        .insert({ release_key: key.padEnd(40, '0'), identity_key: key.padEnd(40, '0'), name: key, source: 'stem', groups: '[]', group_id: 1, files: 1, posted_at: at, readable: false })
+        .returning('id')
+      ids.push(Number(row.id))
+    }
+    // The par2 is held by the first; the second holds the file, and so do 20 releases elsewhere and years apart.
+    const setId = hash('1')
+    await db.table('par2_sets').insert({ set_id: setId, release_id: ids[0], files: 1 })
+    await db.table('par2_files').insert({ set_id: setId, file_id: hash('0'), hash16k: hash('9'), length: 30214, name: 'sample.nfo' })
+    await db.table('file_hashes').insert({ candidate_id: 4000, release_id: ids[1], hash16k: hash('9'), size: 30214 })
+    for (let i = 1; i <= 20; i++) await db.table('file_hashes').insert({ candidate_id: 4000 + i, release_id: 900000 + i, hash16k: hash('9'), size: 30214 })
+    const round = await new Par2Merger().round()
+    assert.deepEqual([round.sets, round.merges], [1, 0])
+    assert.equal(await new Par2Merger().mergeSet(setId), 0)
+    assert.equal((await db.from('releases').whereIn('id', ids).select('id')).length, 2)
+    // One holder fewer: the file is one upload's again.
+    await db.from('file_hashes').where('candidate_id', 4020).delete()
+    assert.equal(await new Par2Merger().mergeSet(setId), 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[] = []