Rebuild: coordinator (planner, NNTP pool, spool, prefetcher, leases, rule sets), node API and the edge:node worker

- groups:add / ranges:plan / nodes:create / coordinator:run commands
- the node API: lease, the range's chunks as one framed stream from the
  spool, batches, heartbeat, finish, release (node guard, token per node)
- edge:node: needs COORDINATOR_URL and NODE_TOKEN only (placeholders for
  the rest), posts a batch every 10 chunks, gives its range back on SIGTERM
- releases are gathered per batch: 1.04 chunk rows per release at 10
  chunks per batch (1.35 when built per XOVER chunk)
- partitions are checked per batch instead of remembered

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

app/controllers/access_tokens_controller.ts

@@ -17,7 +17,7 @@ export default class AccessTokensController {
   }
 
   async destroy({ auth }: HttpContext) {
-    const user = auth.getUserOrFail()
+    const user = auth.use('api').getUserOrFail()
     if (user.currentAccessToken) {
       await User.accessTokens.delete(user, user.currentAccessToken.identifier)
     }

app/controllers/node_controller.ts

@@ -0,0 +1,94 @@
+import type { HttpContext } from '@adonisjs/core/http'
+import { Readable } from 'node:stream'
+import { setTimeout as sleep } from 'node:timers/promises'
+import Leases, { LeaseLost } from '#services/leases'
+import BatchStore, { StaleBatch } from '#services/batch_store'
+import Spool from '#services/spool'
+import { encodeFrame, fromWire, type WireBatch } from '#lib/edge/index'
+
+/** How long the stream waits for a chunk the prefetcher hasn't spooled yet. */
+const CHUNK_WAIT_MS = 15 * 60 * 1000
+
+/**
+ * The edge workers' API (REBUILD.md, parts 1–2): a lease, the range's
+ * chunks as one stream, results per batch, and the end of the range.
+ */
+export default class NodeController {
+  private node(ctx: HttpContext) {
+    return Number(ctx.auth.use('node').getUserOrFail().id)
+  }
+
+  async lease(ctx: HttpContext) {
+    const lease = await Leases.lease(this.node(ctx))
+    if (!lease) return ctx.response.noContent()
+    return lease
+  }
+
+  /** The range's chunks from its resume point, waiting for those still being fetched. */
+  async stream(ctx: HttpContext) {
+    const nodeId = this.node(ctx)
+    const rangeId = Number(ctx.params.id)
+    const range = await this.leased(ctx, rangeId, nodeId)
+    if (!range) return
+    const spans = Spool.chunksOf({ start: Number(range.start_id), end: Number(range.end_id) }, Number(range.next_id))
+    async function* frames() {
+      for (const span of spans) {
+        const waitUntil = Date.now() + CHUNK_WAIT_MS
+        while (!(await Spool.has(rangeId, span))) {
+          if (Date.now() > waitUntil) throw new Error(`Chunk ${span.from}–${span.to} of range ${rangeId} wasn't spooled in time`)
+          await sleep(500)
+        }
+        yield encodeFrame({ ...span, data: await Spool.read(rangeId, span) })
+        // Streaming keeps the lease alive.
+        await Leases.heartbeat(rangeId, nodeId)
+      }
+    }
+    ctx.response.header('Content-Type', 'application/octet-stream')
+    ctx.response.stream(Readable.from(frames()))
+  }
+
+  async batch(ctx: HttpContext) {
+    const nodeId = this.node(ctx)
+    const rangeId = Number(ctx.params.id)
+    try {
+      return await BatchStore.store(rangeId, nodeId, fromWire(ctx.request.body() as WireBatch))
+    } catch (error) {
+      if (error instanceof StaleBatch) return ctx.response.conflict({ error: error.message })
+      throw error
+    }
+  }
+
+  async heartbeat(ctx: HttpContext) {
+    return this.lifecycle(ctx, (rangeId, nodeId) => Leases.heartbeat(rangeId, nodeId))
+  }
+
+  async finish(ctx: HttpContext) {
+    return this.lifecycle(ctx, (rangeId, nodeId) => Leases.finish(rangeId, nodeId))
+  }
+
+  async release(ctx: HttpContext) {
+    return this.lifecycle(ctx, (rangeId, nodeId) => Leases.release(rangeId, nodeId))
+  }
+
+  private async lifecycle(ctx: HttpContext, action: (rangeId: number, nodeId: number) => Promise<void>) {
+    try {
+      await action(Number(ctx.params.id), this.node(ctx))
+      return { ok: true }
+    } catch (error) {
+      if (error instanceof LeaseLost) return ctx.response.conflict({ error: error.message })
+      throw error
+    }
+  }
+
+  private async leased(ctx: HttpContext, rangeId: number, nodeId: number) {
+    try {
+      return await Leases.leased(rangeId, nodeId)
+    } catch (error) {
+      if (error instanceof LeaseLost) {
+        ctx.response.conflict({ error: error.message })
+        return null
+      }
+      throw error
+    }
+  }
+}

app/controllers/profile_controller.ts

@@ -3,6 +3,6 @@ import type { HttpContext } from '@adonisjs/core/http'
 
 export default class ProfileController {
   async show({ auth, serialize }: HttpContext) {
-    return serialize(UserTransformer.transform(auth.getUserOrFail()))
+    return serialize(UserTransformer.transform(auth.use('api').getUserOrFail()))
   }
 }

app/models/node.ts

@@ -0,0 +1,8 @@
+import { NodeSchema } from '#database/schema'
+import { type AccessToken, DbAccessTokensProvider } from '@adonisjs/auth/access_tokens'
+
+/** An edge worker (REBUILD.md, part 2); it authenticates with an access token (the `node` guard). */
+export default class Node extends NodeSchema {
+  static accessTokens = DbAccessTokensProvider.forModel(Node, { table: 'node_access_tokens' })
+  declare currentAccessToken?: AccessToken
+}

app/services/chunk_partitions.ts

@@ -7,15 +7,14 @@ const AHEAD = 2
 /**
  * release_chunks is partitioned by month of writing (REBUILD.md, decision a).
  * Inserts fail without a partition for now(), so the current and the next
- * months' partitions are created before they are needed; checked once per
- * process and month.
+ * months' partitions are created before they are needed.
  */
 class ChunkPartitions {
-  private ensuredFor: string | null = null
-
   async ensure(now = new Date()) {
-    const month = now.toISOString().slice(0, 7)
-    if (this.ensuredFor === month) return
+    // Checked each time, not remembered: a rolled-back transaction can take a created partition along.
+    const current = `release_chunks_${now.toISOString().slice(0, 7).replace('-', '')}`
+    const { rows } = await db.rawQuery(`SELECT to_regclass(?) IS NOT NULL AS present`, [current])
+    if (rows[0].present) return
     const tablespace = env.get('CHUNK_TABLESPACE')
     for (let i = 0; i <= AHEAD; i++) {
       const from = new Date(Date.UTC(now.getUTCFullYear(), now.getUTCMonth() + i, 1))
@@ -25,7 +24,6 @@ class ChunkPartitions {
         `CREATE TABLE IF NOT EXISTS ${name} PARTITION OF release_chunks FOR VALUES FROM ('${from.toISOString()}') TO ('${to.toISOString()}')${tablespace ? ` TABLESPACE ${tablespace}` : ''}`
       )
     }
-    this.ensuredFor = month
   }
 }
 

app/services/leases.ts

@@ -0,0 +1,61 @@
+import db from '@adonisjs/lucid/services/db'
+import coordinatorConfig from '#config/coordinator'
+import RuleSets from '#services/rule_sets'
+import Spool from '#services/spool'
+import type { Lease } from '#lib/edge/index'
+
+export class LeaseLost extends Error {}
+
+/**
+ * Ranges for workers (REBUILD.md, coordinator): the next range whose first
+ * chunk is spooled, by priority (oldest era first); a lease whose worker
+ * went quiet for LEASE_TTL_SECONDS is taken over and resumes at its resume
+ * point.
+ */
+class Leases {
+  async lease(nodeId: number): Promise<Lease | null> {
+    const rules = await RuleSets.current()
+    const { rows } = await db.rawQuery(
+      `UPDATE ranges r SET status = 'leased', node_id = ?, leased_at = now(), heartbeat_at = now(), attempts = attempts + 1, rule_version = ?
+       WHERE r.id = (
+         SELECT id FROM ranges
+         WHERE spooled_chunks > 0 AND (status = 'planned' OR (status = 'leased' AND heartbeat_at < now() - make_interval(secs => ?)))
+         ORDER BY priority, id LIMIT 1 FOR UPDATE SKIP LOCKED)
+       RETURNING r.id, r.group_id, r.start_id, r.end_id, r.next_id`,
+      [nodeId, rules.version, coordinatorConfig.leaseTtlSeconds]
+    )
+    const row = rows[0]
+    if (!row) return null
+    await db.from('nodes').where('id', nodeId).update({ last_seen_at: new Date() })
+    const group = await db.from('groups').where('id', row.group_id).select('name').firstOrFail()
+    return { range: { id: Number(row.id), group: group.name, start: Number(row.start_id), end: Number(row.end_id), next: Number(row.next_id) }, rules }
+  }
+
+  async leased(rangeId: number, nodeId: number) {
+    const range = await db.from('ranges').where('id', rangeId).first()
+    if (!range || range.status !== 'leased' || Number(range.node_id) !== nodeId) throw new LeaseLost(`Range ${rangeId} is not leased to this node`)
+    return range
+  }
+
+  async heartbeat(rangeId: number, nodeId: number) {
+    await this.leased(rangeId, nodeId)
+    await db.from('ranges').where('id', rangeId).update({ heartbeat_at: new Date() })
+    await db.from('nodes').where('id', nodeId).update({ last_seen_at: new Date() })
+  }
+
+  /** The worker reached the range's end: done, and its spooled chunks go. */
+  async finish(rangeId: number, nodeId: number) {
+    const range = await this.leased(rangeId, nodeId)
+    if (Number(range.next_id) <= Number(range.end_id)) throw new LeaseLost(`Range ${rangeId} is at ${range.next_id}, not past its end ${range.end_id}`)
+    await db.from('ranges').where('id', rangeId).update({ status: 'done', done_at: new Date(), heartbeat_at: new Date() })
+    await Spool.remove(rangeId)
+  }
+
+  /** A stopping worker gives its range back; the next one resumes it. */
+  async release(rangeId: number, nodeId: number) {
+    await this.leased(rangeId, nodeId)
+    await db.from('ranges').where('id', rangeId).update({ status: 'planned', node_id: null })
+  }
+}
+
+export default new Leases()

app/services/nntp_pool.ts

@@ -0,0 +1,80 @@
+import env from '#start/env'
+import logger from '@adonisjs/core/services/logger'
+import coordinatorConfig from '#config/coordinator'
+import { XoverClient } from '#lib/nntp/index'
+
+/**
+ * The coordinator's header connections (REBUILD.md, decision e: workers
+ * have no NNTP account). At most NNTP_CONNECTIONS, reused; a connection
+ * that failed is closed and replaced on the next use.
+ */
+class NntpPool {
+  private idle: XoverClient[] = []
+  private open = 0
+  private waiting: ((client: XoverClient) => void)[] = []
+
+  get size() {
+    return coordinatorConfig.nntpConnections
+  }
+
+  async use<T>(work: (client: XoverClient) => Promise<T>): Promise<T> {
+    const client = await this.acquire()
+    let broken = false
+    try {
+      return await work(client)
+    } catch (error) {
+      broken = true
+      throw error
+    } finally {
+      this.release(client, broken)
+    }
+  }
+
+  private async acquire(): Promise<XoverClient> {
+    const ready = this.idle.pop()
+    if (ready) return ready
+    if (this.open < this.size) {
+      this.open++
+      try {
+        return await XoverClient.connect({
+          host: env.get('NNTP_HOST')!,
+          port: env.get('NNTP_PORT', 563),
+          user: env.get('NNTP_USER'),
+          password: env.get('NNTP_PASSWORD'),
+          secure: env.get('NNTP_SECURE', true),
+          compress: true,
+          timeoutMs: coordinatorConfig.nntpTimeoutMs,
+        })
+      } catch (error) {
+        this.open--
+        throw error
+      }
+    }
+    return new Promise((resolve) => this.waiting.push(resolve))
+  }
+
+  private release(client: XoverClient, broken: boolean) {
+    if (broken) {
+      this.open--
+      try {
+        client.close()
+      } catch (error: any) {
+        logger.debug(`Closing a failed NNTP connection: ${error.message}`)
+      }
+      // Someone waiting gets a fresh connection.
+      const next = this.waiting.shift()
+      if (next) this.acquire().then(next, () => this.waiting.unshift(next))
+      return
+    }
+    const next = this.waiting.shift()
+    if (next) next(client)
+    else this.idle.push(client)
+  }
+
+  async close() {
+    for (const client of this.idle.splice(0)) client.close()
+    this.open = 0
+  }
+}
+
+export default new NntpPool()

app/services/planner.ts

@@ -0,0 +1,64 @@
+import db from '@adonisjs/lucid/services/db'
+import coordinatorConfig from '#config/coordinator'
+import NntpPool from '#services/nntp_pool'
+
+/**
+ * The indexed groups and their ranges (REBUILD.md, group list, coordinator).
+ * New groups get priorities after the existing ones (owner rule). A group's
+ * span from GROUP is cut into ranges of RANGE_ARTICLES; a range's priority is
+ * its position in its group's history (0 oldest … 1000 newest), so all groups
+ * start with their oldest era. Planning again only adds ranges for new articles.
+ */
+class Planner {
+  async addGroups(names: string[]): Promise<number> {
+    let added = 0
+    for (const name of names) {
+      const { rows } = await db.rawQuery(
+        `INSERT INTO groups (name, priority) SELECT ?, coalesce(max(priority), 0) + 1 FROM groups ON CONFLICT (name) DO NOTHING RETURNING id`,
+        [name]
+      )
+      added += rows.length
+    }
+    return added
+  }
+
+  /** The `top` groups matching `pattern` by article span, largest first (LIST ACTIVE). */
+  async largestGroups(pattern: string, top: number): Promise<string[]> {
+    const groups = await NntpPool.use((client) => client.listActive(pattern))
+    return groups
+      .filter((g) => g.high > g.low)
+      .sort((a, b) => b.high - b.low - (a.high - a.low))
+      .slice(0, top)
+      .map((g) => g.name)
+  }
+
+  async planGroup(group: { id: number; name: string }): Promise<number> {
+    const info = await NntpPool.use((client) => client.group(group.name))
+    const size = coordinatorConfig.rangeArticles
+    const last = await db.from('ranges').where('group_id', group.id).max('end_id as end').first()
+    const from = Math.max(info.low, Number(last?.end ?? 0) + 1)
+    const span = Math.max(1, info.high - info.low)
+    const rows = []
+    for (let start = from; start <= info.high; start += size) {
+      rows.push({
+        group_id: group.id,
+        start_id: start,
+        end_id: Math.min(start + size - 1, info.high),
+        next_id: start,
+        priority: Math.floor((1000 * (start - info.low)) / span),
+      })
+    }
+    for (let i = 0; i < rows.length; i += 2000) await db.table('ranges').multiInsert(rows.slice(i, i + 2000))
+    await db.from('groups').where('id', group.id).update({ low: info.low, high: info.high, planned_at: new Date() })
+    return rows.length
+  }
+
+  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 })
+    return { groups: groups.length, ranges }
+  }
+}
+
+export default new Planner()

app/services/prefetcher.ts

@@ -0,0 +1,90 @@
+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'
+import Spool from '#services/spool'
+
+interface Target {
+  id: number
+  group: string
+  start: number
+  end: number
+  next: number
+  leased: boolean
+}
+
+/**
+ * Keeps the news server busy ahead of the workers (REBUILD.md, prefetch):
+ * the missing chunks of leased ranges first, then the next planned ranges,
+ * about SPOOL_AHEAD_PER_LEASE per active lease (at least SPOOL_MIN_AHEAD),
+ * within SPOOL_MAX_GB. One range per connection at a time, in order.
+ */
+export class Prefetcher {
+  private busy = new Set<number>()
+  private stopped = false
+
+  /** One round: starts work for free connections; returns how many ranges it started. */
+  async tick(): Promise<number> {
+    const free = NntpPool.size - this.busy.size
+    if (free <= 0) return 0
+    const targets = await this.targets(free)
+    for (const target of targets) {
+      this.busy.add(target.id)
+      this.spoolRange(target)
+        .catch((error) => logger.warn(`Spooling range ${target.id}: ${error.message}`))
+        .finally(() => this.busy.delete(target.id))
+    }
+    return targets.length
+  }
+
+  stop() {
+    this.stopped = true
+  }
+
+  get active() {
+    return this.busy.size
+  }
+
+  private async targets(limit: number): Promise<Target[]> {
+    const busy = [...this.busy]
+    const select = (q: any) =>
+      q
+        .from('ranges as r')
+        .join('groups as g', 'g.id', 'r.group_id')
+        .whereNull('r.spooled_at')
+        .whereNotIn('r.id', busy.length > 0 ? busy : [0])
+        .select('r.id', 'g.name as group', 'r.start_id', 'r.end_id', 'r.next_id', 'r.status')
+    // Leased ranges whose chunks aren't all there: a worker is waiting for them.
+    const leased = await select(db.query()).where('r.status', 'leased').orderBy('r.leased_at').limit(limit)
+    const out: Target[] = leased.map(toTarget)
+    if (out.length >= limit) return out
+
+    const [{ leases }] = await db.from('ranges').where('status', 'leased').count('* as leases')
+    const [{ ahead }] = await db.from('ranges').where('status', 'planned').where('spooled_chunks', '>', 0).count('* as ahead')
+    const wanted = Math.max(coordinatorConfig.spool.minAhead, coordinatorConfig.spool.aheadPerLease * Number(leases)) - Number(ahead)
+    if (wanted <= 0 || (await Spool.size()) >= coordinatorConfig.spool.maxBytes) return out
+    const planned = await select(db.query()).where('r.status', 'planned').orderBy([{ column: 'r.priority' }, { column: 'r.id' }]).limit(Math.min(limit - out.length, wanted))
+    return [...out, ...planned.map(toTarget)]
+  }
+
+  private async spoolRange(target: Target) {
+    let count = 0
+    for (const span of Spool.chunksOf({ start: target.start, end: target.end }, target.next)) {
+      if (this.stopped) return
+      if (!(await Spool.has(target.id, span))) {
+        const data = await NntpPool.use(async (client) => {
+          await client.group(target.group)
+          return client.xoverRaw(span.from, span.to)
+        })
+        await Spool.write(target.id, span, data)
+      }
+      count++
+      await db.from('ranges').where('id', target.id).update({ spooled_chunks: count })
+    }
+    await db.from('ranges').where('id', target.id).update({ spooled_at: new Date() })
+  }
+}
+
+function toTarget(row: any): Target {
+  return { id: Number(row.id), group: row.group, start: Number(row.start_id), end: Number(row.end_id), next: Number(row.next_id), leased: row.status === 'leased' }
+}

app/services/rule_sets.ts

@@ -0,0 +1,36 @@
+import db from '@adonisjs/lucid/services/db'
+import { DEFAULT_RULES, type Rules } from '#lib/rules/index'
+
+type StoredRules = Omit<Rules, 'version' | 'priorities'>
+
+/**
+ * The rule set workers get with a lease (REBUILD.md, rules and recovery):
+ * the newest version, plus the indexed groups' priorities (owner rule).
+ * Version 1 (DEFAULT_RULES, no dropped signatures) is created when there is none.
+ */
+class RuleSets {
+  async current(): Promise<Rules> {
+    let row = await db.from('rule_sets').orderBy('version', 'desc').first()
+    if (!row) {
+      const { version: _v, ...rules } = DEFAULT_RULES
+      await db.table('rule_sets').insert({ version: 1, rules: JSON.stringify({ ...rules, droppedSignatures: [] }), note: 'default' }).onConflict('version').ignore()
+      row = await db.from('rule_sets').orderBy('version', 'desc').first()
+    }
+    const groups = await db.from('groups').where('active', true).select('name', 'priority')
+    return {
+      ...(row.rules as StoredRules),
+      version: Number(row.version),
+      priorities: Object.fromEntries(groups.map((g) => [g.name, Number(g.priority)])),
+    }
+  }
+
+  /** A new version: what changes is recorded on the ranges that run with it. */
+  async publish(rules: StoredRules, note?: string): Promise<number> {
+    const latest = await db.from('rule_sets').max('version as v').first()
+    const version = Number(latest?.v ?? 0) + 1
+    await db.table('rule_sets').insert({ version, rules: JSON.stringify(rules), note: note ?? null })
+    return version
+  }
+}
+
+export default new RuleSets()

app/services/spool.ts

@@ -0,0 +1,78 @@
+import { mkdir, readFile, readdir, rename, rm, stat, writeFile } from 'node:fs/promises'
+import { join } from 'node:path'
+import coordinatorConfig from '#config/coordinator'
+
+export interface ChunkSpan {
+  from: number
+  to: number
+}
+
+/**
+ * Prefetched XOVER chunks on disk (REBUILD.md, prefetch): `<dir>/<range id>/<from>-<to>.gz`,
+ * each as the news server sent it. Chunks follow a fixed grid from the
+ * range's start, so a resume point (always at a batch boundary) is a chunk's start.
+ */
+class Spool {
+  get dir() {
+    return coordinatorConfig.spool.dir
+  }
+
+  /** The chunks of a range from `from` (its resume point) to its end. */
+  chunksOf(range: { start: number; end: number }, from = range.start): ChunkSpan[] {
+    const size = coordinatorConfig.chunkArticles
+    const spans: ChunkSpan[] = []
+    for (let at = range.start; at <= range.end; at += size) {
+      const span = { from: at, to: Math.min(at + size - 1, range.end) }
+      if (span.to >= from) spans.push(span)
+    }
+    return spans
+  }
+
+  private path(rangeId: number, span: ChunkSpan) {
+    return join(this.dir, String(rangeId), `${span.from}-${span.to}.gz`)
+  }
+
+  async has(rangeId: number, span: ChunkSpan) {
+    try {
+      await stat(this.path(rangeId, span))
+      return true
+    } catch {
+      return false
+    }
+  }
+
+  async write(rangeId: number, span: ChunkSpan, data: Buffer) {
+    const file = this.path(rangeId, span)
+    await mkdir(join(this.dir, String(rangeId)), { recursive: true })
+    // Written under another name first: a reader never sees half a chunk.
+    await writeFile(`${file}.part`, data)
+    await rename(`${file}.part`, file)
+  }
+
+  read(rangeId: number, span: ChunkSpan) {
+    return readFile(this.path(rangeId, span))
+  }
+
+  async remove(rangeId: number) {
+    await rm(join(this.dir, String(rangeId)), { recursive: true, force: true })
+  }
+
+  /** Bytes in the spool. */
+  async size(): Promise<number> {
+    let total = 0
+    let ranges: string[] = []
+    try {
+      ranges = await readdir(this.dir)
+    } catch {
+      return 0
+    }
+    for (const range of ranges) {
+      for (const file of await readdir(join(this.dir, range)).catch(() => [])) {
+        total += (await stat(join(this.dir, range, file)).catch(() => ({ size: 0 }))).size
+      }
+    }
+    return total
+  }
+}
+
+export default new Spool()