Rebuild: the fragment merger

Files cut by a range's edge are put together once the ranges around them are done (coordinator:run, every minute): a complete file is judged by the rules and written as a release chunk with probe candidates; one still missing parts while both neighbouring ranges are done is dropped and counted (incomplete/edge). A neighbour not planned yet counts as not done, except before the group's oldest article. BatchStore.writeReleases is the shared write path. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
2026-10-09 02:19:53 +02:00
1 parent 2c14315f94
commit b862201991
@@ -1,5 +1,6 @@
 import db from '@adonisjs/lucid/services/db'
-import type { EdgeBatch } from '#lib/edge/index'
+import type { TransactionClientContract } from '@adonisjs/lucid/types/database'
+import type { EdgeBatch, ProbeCandidate, ReleaseChunk } from '#lib/edge/index'
 import ChunkPartitions from '#services/chunk_partitions'

 export class StaleBatch extends Error {}
@@ -23,66 +24,28 @@ class BatchStore {
    await ChunkPartitions.ensure()
    return db.transaction(async (trx) => {
      const range = await trx.from('ranges').where('id', rangeId).forUpdate().first()
-      if (!range || range.status !== 'leased' || Number(range.node_id) !== nodeId) throw new StaleBatch(`Range ${rangeId} is not leased to this node`)
-      if (Number(range.next_id) !== batch.from) throw new StaleBatch(`Range ${rangeId} is at ${range.next_id}, the batch starts at ${batch.from}`)
-      const result: StoredBatch = { releases: batch.releases.length, newReleases: 0, candidates: 0, fragments: batch.fragments.length }
-
-      // Releases, sorted by key so concurrent batches lock them in the same order.
-      const releases = [...batch.releases].sort((a, b) => a.key.localeCompare(b.key))
-      const ids = new Map<string, number>()
-      for (let i = 0; i < releases.length; i += 1000) {
-        const chunk = releases.slice(i, i + 1000)
-        const { rows } = await trx.rawQuery(
-          `INSERT INTO releases (release_key, name, source, poster, groups, posted_at, file_total, files, segments, bytes, chunks, first_article, last_article, readable, has_par2)
-           SELECT v.key, v.name, v.source, v.poster, v.groups, to_timestamp(v.posted_at / 1000.0), v.file_total, v.files, v.segments, v.bytes, 1, v.first_article, v.last_article, v.readable, v.has_par2
-           FROM jsonb_to_recordset(?::jsonb) AS v(key text, name text, source text, poster text, groups jsonb, posted_at bigint, file_total int, files int, segments bigint, bytes bigint, first_article bigint, last_article bigint, readable boolean, has_par2 boolean)
-           ON CONFLICT (release_key) DO UPDATE SET
-             files = releases.files + excluded.files,
-             segments = releases.segments + excluded.segments,
-             bytes = releases.bytes + excluded.bytes,
-             chunks = releases.chunks + 1,
-             posted_at = least(releases.posted_at, excluded.posted_at),
-             file_total = coalesce(releases.file_total, excluded.file_total),
-             first_article = least(releases.first_article, excluded.first_article),
-             last_article = greatest(releases.last_article, excluded.last_article),
-             readable = releases.readable OR excluded.readable,
-             has_par2 = releases.has_par2 OR excluded.has_par2,
-             last_seen_at = now()
-           RETURNING id, release_key, (xmax = 0) AS inserted`,
-          [
-            JSON.stringify(
-              chunk.map((r) => ({
-                key: r.key, name: r.name, source: r.source, poster: r.poster, groups: r.groups, posted_at: r.postedAt, file_total: r.fileTotal,
-                files: r.files, segments: r.segments, bytes: r.bytes, first_article: r.firstArticle, last_article: r.lastArticle, readable: r.readable, has_par2: r.hasPar2,
-              }))
-            ),
-          ]
+      if (!range || range.status !== 'leased' || Number(range.node_id) !== nodeId)
+        throw new StaleBatch(`Range ${rangeId} is not leased to this node`)
+      if (Number(range.next_id) !== batch.from)
+        throw new StaleBatch(
+          `Range ${rangeId} is at ${range.next_id}, the batch starts at ${batch.from}`
        )
-        for (const row of rows) {
-          ids.set(row.release_key, Number(row.id))
-          if (row.inserted) result.newReleases++
-        }
+      const result: StoredBatch = {
+        releases: batch.releases.length,
+        newReleases: 0,
+        candidates: 0,
+        fragments: batch.fragments.length,
      }

-      // One write-once chunk per release.
-      for (let i = 0; i < releases.length; i += 500) {
-        const rows = releases.slice(i, i + 500).map((r) => ({
-          release_id: ids.get(r.key)!,
-          range_id: rangeId,
-          batch_from: batch.from,
-          files: r.files,
-          segments: r.segments,
-          payload: r.payload,
-        }))
-        if (rows.length > 0) await trx.insertQuery().table('release_chunks').multiInsert(rows)
-      }
-
-      const candidates = batch.candidates.filter((c) => ids.has(c.releaseKey))
-      for (let i = 0; i < candidates.length; i += 2000) {
-        const rows = candidates.slice(i, i + 2000).map((c) => ({ release_id: ids.get(c.releaseKey)!, kind: c.kind, message_id: c.messageId, file: c.file }))
-        const added = await trx.knexQuery().table('probe_candidates').insert(rows).onConflict(['release_id', 'kind', 'message_id']).ignore().returning('id')
-        result.candidates += added.length
-      }
+      const written = await this.writeReleases(
+        trx,
+        rangeId,
+        batch.from,
+        batch.releases,
+        batch.candidates
+      )
+      result.newReleases = written.newReleases
+      result.candidates = written.candidates

      if (batch.counts.length > 0) {
        await trx.rawQuery(
@@ -95,7 +58,14 @@ class BatchStore {

      for (const s of batch.samples) {
        await trx.insertQuery().table('rule_samples').insert({
-          range_id: rangeId, rule: s.rule, rule_key: s.ruleKey, release_key: s.key, name: s.name, files: s.files, segments: s.segments, payload: s.payload,
+          range_id: rangeId,
+          rule: s.rule,
+          rule_key: s.ruleKey,
+          release_key: s.key,
+          name: s.name,
+          files: s.files,
+          segments: s.segments,
+          payload: s.payload,
        })
      }

@@ -113,10 +83,112 @@ class BatchStore {
      await trx
        .from('ranges')
        .where('id', rangeId)
-        .update({ next_id: batch.next, headers: db.raw('headers + ?', [batch.headers]), heartbeat_at: new Date() })
+        .update({
+          next_id: batch.next,
+          headers: db.raw('headers + ?', [batch.headers]),
+          heartbeat_at: new Date(),
+        })
      return result
    })
  }
+
+  /**
+   * Release aggregates in the catalog, one write-once chunk per release, and
+   * probe candidates (a batch's, or files the fragment merger completed).
+   * Releases are sorted by key, so concurrent writers lock them in the same order.
+   */
+  async writeReleases(
+    trx: TransactionClientContract,
+    rangeId: number,
+    batchFrom: number,
+    releases: ReleaseChunk[],
+    candidates: ProbeCandidate[]
+  ) {
+    const out = { newReleases: 0, candidates: 0, ids: new Map<string, number>() }
+    const ids = out.ids
+    // Releases, sorted by key so concurrent batches lock them in the same order.
+    const sorted = [...releases].sort((a, b) => a.key.localeCompare(b.key))
+    for (let i = 0; i < sorted.length; i += 1000) {
+      const chunk = sorted.slice(i, i + 1000)
+      const { rows } = await trx.rawQuery(
+        `INSERT INTO releases (release_key, name, source, poster, groups, posted_at, file_total, files, segments, bytes, chunks, first_article, last_article, readable, has_par2)
+           SELECT v.key, v.name, v.source, v.poster, v.groups, to_timestamp(v.posted_at / 1000.0), v.file_total, v.files, v.segments, v.bytes, 1, v.first_article, v.last_article, v.readable, v.has_par2
+           FROM jsonb_to_recordset(?::jsonb) AS v(key text, name text, source text, poster text, groups jsonb, posted_at bigint, file_total int, files int, segments bigint, bytes bigint, first_article bigint, last_article bigint, readable boolean, has_par2 boolean)
+           ON CONFLICT (release_key) DO UPDATE SET
+             files = releases.files + excluded.files,
+             segments = releases.segments + excluded.segments,
+             bytes = releases.bytes + excluded.bytes,
+             chunks = releases.chunks + 1,
+             posted_at = least(releases.posted_at, excluded.posted_at),
+             file_total = coalesce(releases.file_total, excluded.file_total),
+             first_article = least(releases.first_article, excluded.first_article),
+             last_article = greatest(releases.last_article, excluded.last_article),
+             readable = releases.readable OR excluded.readable,
+             has_par2 = releases.has_par2 OR excluded.has_par2,
+             last_seen_at = now()
+           RETURNING id, release_key, (xmax = 0) AS inserted`,
+        [
+          JSON.stringify(
+            chunk.map((r) => ({
+              key: r.key,
+              name: r.name,
+              source: r.source,
+              poster: r.poster,
+              groups: r.groups,
+              posted_at: r.postedAt,
+              file_total: r.fileTotal,
+              files: r.files,
+              segments: r.segments,
+              bytes: r.bytes,
+              first_article: r.firstArticle,
+              last_article: r.lastArticle,
+              readable: r.readable,
+              has_par2: r.hasPar2,
+            }))
+          ),
+        ]
+      )
+      for (const row of rows) {
+        ids.set(row.release_key, Number(row.id))
+        if (row.inserted) out.newReleases++
+      }
+    }
+
+    // One write-once chunk per release.
+    for (let i = 0; i < sorted.length; i += 500) {
+      const rows = sorted.slice(i, i + 500).map((r) => ({
+        release_id: ids.get(r.key)!,
+        range_id: rangeId,
+        batch_from: batchFrom,
+        files: r.files,
+        segments: r.segments,
+        payload: r.payload,
+      }))
+      if (rows.length > 0) await trx.insertQuery().table('release_chunks').multiInsert(rows)
+    }
+
+    const kept = candidates.filter((c) => ids.has(c.releaseKey))
+    for (let i = 0; i < kept.length; i += 2000) {
+      const rows = kept
+        .slice(i, i + 2000)
+        .map((c) => ({
+          release_id: ids.get(c.releaseKey)!,
+          kind: c.kind,
+          message_id: c.messageId,
+          file: c.file,
+        }))
+      const added = await trx
+        .knexQuery()
+        .table('probe_candidates')
+        .insert(rows)
+        .onConflict(['release_id', 'kind', 'message_id'])
+        .ignore()
+        .returning('id')
+      out.candidates += added.length
+    }
+
+    return out
+  }
 }

 export default new BatchStore()