Rebuild: merger rounds with one neighbour query, a cursor and a time budget; planning skips failing groups

The neighbour check was three queries per file (~2,000 files/minute); now
one per round of 20k files, with a cursor so waiting files don't take
every round's places, and rounds until nothing is left or 30 s. On the
dev backlog (old-style fragments): ~18k files/minute in the 30 s budget;
at full throughput the new fragment rule leaves ~1,200 files/minute.
A group that fails GROUP (gone from the server) no longer stops planning
the others; edge:bench's groups are inactive.

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

app/services/fragment_merger.ts

@@ -15,7 +15,7 @@ export interface MergeResult {
 }
 
 /** File keys per round. */
-const BATCH = 2000
+const BATCH = 20_000
 
 /**
  * Files cut by a range's edge (REBUILD.md, fragments), put together once
@@ -25,12 +25,18 @@ const BATCH = 2000
  * incomplete, and dropped (counted on its ranges). Others wait.
  */
 class FragmentMerger {
+  /** Where the next round goes on: waiting files must not take every round's places. */
+  private cursor = ''
+
   async run(): Promise<MergeResult> {
     const result: MergeResult = { files: 0, completed: 0, incomplete: 0, waiting: 0 }
     const { rows: keys } = await db.rawQuery(
-      `SELECT DISTINCT f.file_key FROM fragments f JOIN ranges r ON r.id = f.range_id WHERE r.status = 'done' LIMIT ?`,
-      [BATCH]
+      `SELECT DISTINCT f.file_key FROM fragments f JOIN ranges r ON r.id = f.range_id
+       WHERE r.status = 'done' AND f.file_key > ? ORDER BY f.file_key LIMIT ?`,
+      [this.cursor, BATCH]
     )
+    // At the end of the keys: the next round starts over.
+    this.cursor = keys.length < BATCH ? '' : keys.at(-1).file_key
     if (keys.length === 0) return result
     const rules = new RuleSet(await RuleSets.current())
     const rows = await db
@@ -47,6 +53,7 @@ class FragmentMerger {
 
     const completed: { file: AggregatedFile<EdgeSegment>; rangeId: number; ids: number[] }[] = []
     const incomplete: { rangeId: number; ids: number[]; segments: number }[] = []
+    const neighbours = await this.neighbourStatus([...byKey.values()])
     for (const list of byKey.values()) {
       result.files++
       const file = mergeFragments(list.map((r) => r.fragment as Fragment))
@@ -56,7 +63,7 @@ class FragmentMerger {
         completed.push({ file, rangeId, ids })
         continue
       }
-      if (list.every((r) => r.status === 'done') && (await this.neighboursDone(list))) {
+      if (list.every((r) => r.status === 'done') && neighbours(list)) {
         incomplete.push({ rangeId, ids, segments: list.reduce((n, r) => n + (r.fragment as Fragment).segments.length, 0) })
       } else {
         result.waiting++
@@ -96,21 +103,40 @@ class FragmentMerger {
   }
 
   /**
-   * Whether the ranges right before and after these fragments' ranges (where
-   * a missing part could be) are done. One that isn't planned yet counts as
-   * not done (newer articles come later), except before the group's oldest article.
+   * For each file's fragments: whether the ranges right before and after
+   * their ranges (where a missing part could be) are done, read in one
+   * query for the whole round (per file it was three, ~2,000 files a
+   * minute). A neighbour not planned yet counts as not done (newer articles
+   * come later), except before the group's oldest article.
    */
-  private async neighboursDone(list: { group_id: number; start_id: number | string; end_id: number | string }[]) {
-    const groupId = list[0].group_id
-    const first = Math.min(...list.map((r) => Number(r.start_id)))
-    const last = Math.max(...list.map((r) => Number(r.end_id)))
-    const [before, after, group] = await Promise.all([
-      db.from('ranges').where('group_id', groupId).where('end_id', first - 1).select('status').first(),
-      db.from('ranges').where('group_id', groupId).where('start_id', last + 1).select('status').first(),
-      db.from('groups').where('id', groupId).select('low').first(),
-    ])
-    const beforeDone = before ? before.status === 'done' : group?.low === null || first <= Number(group?.low ?? 0)
-    return beforeDone && after?.status === 'done'
+  private async neighbourStatus(lists: { group_id: number; start_id: number | string; end_id: number | string }[][]) {
+    const span = (list: (typeof lists)[number]) => ({
+      group: Number(list[0].group_id),
+      first: Math.min(...list.map((r) => Number(r.start_id))),
+      last: Math.max(...list.map((r) => Number(r.end_id))),
+    })
+    const spans = lists.map(span)
+    const groups = [...new Set(spans.map((s) => s.group))]
+    const { rows } = await db.rawQuery(
+      `SELECT r.group_id, r.start_id, r.end_id, r.status FROM ranges r
+       JOIN unnest(?::int[], ?::bigint[], ?::bigint[]) AS s(group_id, before_end, after_start)
+         ON r.group_id = s.group_id AND (r.end_id = s.before_end OR r.start_id = s.after_start)`,
+      [spans.map((s) => s.group), spans.map((s) => s.first - 1), spans.map((s) => s.last + 1)]
+    )
+    const byEnd = new Map<string, string>()
+    const byStart = new Map<string, string>()
+    for (const r of rows) {
+      byEnd.set(`${r.group_id}:${r.end_id}`, r.status)
+      byStart.set(`${r.group_id}:${r.start_id}`, r.status)
+    }
+    const lows = new Map((await db.from('groups').whereIn('id', groups).select('id', 'low')).map((g) => [Number(g.id), g.low === null ? null : Number(g.low)]))
+    return (list: (typeof lists)[number]) => {
+      const { group, first, last } = span(list)
+      const before = byEnd.get(`${group}:${first - 1}`)
+      const low = lows.get(group)
+      const beforeDone = before !== undefined ? before === 'done' : low === null || low === undefined || first <= low
+      return beforeDone && byStart.get(`${group}:${last + 1}`) === 'done'
+    }
   }
 }
 

app/services/planner.ts

@@ -1,4 +1,5 @@
 import db from '@adonisjs/lucid/services/db'
+import logger from '@adonisjs/core/services/logger'
 import coordinatorConfig from '#config/coordinator'
 import NntpPool from '#services/nntp_pool'
 
@@ -56,7 +57,14 @@ class Planner {
   async planAll(): Promise<{ groups: number; ranges: number }> {
     const groups = await db.from('groups').where('active', true).orderBy('priority').select('id', 'name')
     let ranges = 0
-    for (const group of groups) ranges += await this.planGroup({ id: Number(group.id), name: group.name })
+    for (const group of groups) {
+      // One group failing (e.g. gone from the server) doesn't stop the others.
+      try {
+        ranges += await this.planGroup({ id: Number(group.id), name: group.name })
+      } catch (error: any) {
+        logger.warn(`Planning ${group.name}: ${error.message}`)
+      }
+    }
     return { groups: groups.length, ranges }
   }
 }

commands/coordinator_run.ts

@@ -9,6 +9,7 @@ import NntpPool from '#services/nntp_pool'
 const TICK_MS = 2000
 const PLAN_EVERY_MS = 60 * 60 * 1000
 const MERGE_EVERY_MS = 60 * 1000
+const MERGE_BUDGET_MS = 30 * 1000
 
 /**
  * The coordinator's background work (REBUILD.md, part 1): the prefetcher,
@@ -44,8 +45,16 @@ export default class CoordinatorRun extends BaseCommand {
         }
         if (Date.now() - mergedAt >= MERGE_EVERY_MS) {
           mergedAt = Date.now()
-          const merged = await FragmentMerger.run()
-          if (merged.completed || merged.incomplete) this.logger.info(`Fragments: ${merged.completed} files completed, ${merged.incomplete} incomplete, ${merged.waiting} waiting.`)
+          // Rounds until nothing more can be decided, or MERGE_BUDGET_MS.
+          const totals = { completed: 0, incomplete: 0, waiting: 0 }
+          while (!stopping && Date.now() - mergedAt < MERGE_BUDGET_MS) {
+            const merged = await FragmentMerger.run()
+            totals.completed += merged.completed
+            totals.incomplete += merged.incomplete
+            totals.waiting = merged.waiting
+            if (merged.files === 0) break
+          }
+          if (totals.completed || totals.incomplete) this.logger.info(`Fragments: ${totals.completed} files completed, ${totals.incomplete} incomplete, ${totals.waiting} waiting.`)
         }
         await prefetcher.tick()
       } catch (error: any) {

commands/edge_bench.ts

@@ -101,7 +101,8 @@ export default class EdgeBench extends BaseCommand {
     let rangeId = 0
     let nodeId = 0
     if (this.store) {
-      const [g] = await db.table('groups').insert({ name: `bench:${group}:${Date.now()}`, priority: 1_000_000 + (Date.now() % 1_000_000) }).returning('id')
+      // Inactive: the coordinator neither plans nor leases it.
+      const [g] = await db.table('groups').insert({ name: `bench:${group}:${Date.now()}`, priority: 1_000_000 + (Date.now() % 1_000_000), active: false }).returning('id')
       const [n] = await db.table('nodes').insert({ name: `bench-${Date.now()}` }).returning('id')
       nodeId = Number(n.id)
       const [r] = await db.table('ranges').insert({ group_id: g.id, start_id: start, end_id: end, next_id: start, status: 'leased', node_id: nodeId }).returning('id')