Seed orchestrator: remote nodes lease ranges and post results over HTTPS

- seed_nodes (hashed bearer tokens) and seed_ranges (planned ranges with
  lease, resume point and counts); migration …140.
- SeedOrchestrator: lease (oldest plan first, FOR UPDATE SKIP LOCKED; a
  lease without a batch for 15 min passes to the next node and resumes at
  next_id), batch (stores files and fragments, moves next_id), finish,
  release, plan (from below live indexing / the backfill / earlier plans
  down to N days back; takes over the group's backfill), status.
- Seed API under /api/v1/seed (lease, ranges/:id/batches|done|release,
  status) behind a node-token middleware; JSON bodies up to 64 MB after
  gzip.
- seed:node runs a node: no database or Redis, outgoing HTTPS only, N
  connections, gzip batches, retries for network errors and 5xx, drops a
  range on 409, gives its ranges back on SIGTERM. seed-node/ has a compose
  file that builds the image from this repository.
- seed:token, seed:plan, seed:status for the operator. Header parsing moved
  to lib/ingest and the range loop to lib/seed/runner, shared by seed:range
  and seed:node.

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

commands/SeedNode.ts

@@ -0,0 +1,194 @@
+import { gzipSync } from 'node:zlib'
+import { setTimeout as sleep } from 'node:timers/promises'
+import { BaseCommand, flags } from '@adonisjs/core/ace'
+import { CommandOptions } from '@adonisjs/core/types/ace'
+import nntpConfig from '#config/nntp'
+import indexerConfig from '#config/indexer'
+import { XoverClient } from '#lib/nntp/xover_client'
+import { runRange } from '#lib/seed/runner'
+import type { SeedBatch } from '#lib/seed/wire'
+import type { OverviewHeader } from '#lib/ingest/segment'
+
+class LeaseLost extends Error {}
+class Stopping extends Error {}
+
+interface Lease {
+  id: number
+  group: string
+  start: number
+  end: number
+}
+
+/**
+ * A seeding node (CONCEPT.md D38): leases article ranges from the seed API
+ * over HTTPS, fetches and aggregates their headers over its own NNTP
+ * connection(s) (D37), and posts the results back in gzip-compressed
+ * batches. Outgoing requests only: no ports, database or Redis needed.
+ *
+ *   SEED_URL=https://seed.example SEED_TOKEN=sn_… NNTP_HOST=… node ace seed:node --connections 4
+ */
+export default class SeedNode extends BaseCommand {
+  public static commandName = 'seed:node'
+  public static description = 'Runs a seeding node: leases ranges from the seed API and submits what it aggregated.'
+
+  public static options: CommandOptions = {
+    startApp: false,
+    staysAlive: true,
+  }
+
+  @flags.string({ description: 'Seed API base URL (default: SEED_URL)' })
+  declare url?: string
+
+  @flags.string({ description: 'Node token (default: SEED_TOKEN)' })
+  declare token?: string
+
+  @flags.number({ description: 'Ranges worked on at once, one NNTP connection each (default: SEED_CONNECTIONS or 1)', default: Number(process.env.SEED_CONNECTIONS) || 1 })
+  declare connections: number
+
+  @flags.number({ description: 'Articles per XOVER', default: 20_000 })
+  declare chunk: number
+
+  @flags.number({ description: 'XOVER chunks per batch posted to the API', default: 5 })
+  declare submitEvery: number
+
+  @flags.boolean({ description: 'Stop when there is nothing to lease instead of waiting' })
+  declare once: boolean
+
+  private stopping = false
+
+  public async run() {
+    const base = (this.url ?? process.env.SEED_URL ?? '').replace(/\/+$/, '')
+    const token = this.token ?? process.env.SEED_TOKEN ?? ''
+    if (!base || !token) {
+      this.logger.error('Set --url and --token (or SEED_URL and SEED_TOKEN).')
+      this.exitCode = 1
+      return
+    }
+    const stop = () => {
+      if (!this.stopping) this.logger.info('Stopping: giving back the ranges in progress…')
+      this.stopping = true
+    }
+    process.on('SIGTERM', stop)
+    process.on('SIGINT', stop)
+
+    const api = new SeedApi(base, token, (message) => this.logger.warning(message))
+    this.logger.info(`Seeding node for ${base} with ${this.connections} connection(s).`)
+    await Promise.all(Array.from({ length: this.connections }, (_, i) => this.work(api, i + 1)))
+    this.logger.info('Seeding node stopped.')
+  }
+
+  private async work(api: SeedApi, worker: number) {
+    const log = (message: string) => this.logger.info(`[${worker}] ${message}`)
+    while (!this.stopping) {
+      let lease: Lease | null
+      try {
+        lease = ((await api.post('/api/v1/seed/lease')) as { range: Lease } | null)?.range ?? null
+      } catch (error: any) {
+        if (error instanceof Unauthorized) throw error
+        log(`Lease failed: ${error.message}`)
+        await this.pause(60_000)
+        continue
+      }
+      if (!lease) {
+        if (this.once) return
+        await this.pause(60_000)
+        continue
+      }
+
+      log(`${lease.group} ${lease.start}–${lease.end}`)
+      const started = Date.now()
+      let headers = 0
+      let client: XoverClient | undefined
+      try {
+        client = await XoverClient.connect({
+          host: nntpConfig.host,
+          port: nntpConfig.port,
+          secure: nntpConfig.secure,
+          user: nntpConfig.user,
+          password: nntpConfig.password?.release(),
+          timeoutMs: nntpConfig.commandTimeoutMs,
+          compress: true,
+        })
+        await runRange({
+          client,
+          group: lease.group,
+          start: lease.start,
+          end: lease.end,
+          chunk: this.chunk,
+          submitEvery: this.submitEvery,
+          isDeferred: (header: OverviewHeader) => indexerConfig.deferBodyMessageIds.some((re) => re.test(header['message-id'] ?? '')),
+          sink: async (batch: SeedBatch) => {
+            if (this.stopping) throw new Stopping()
+            await api.post(`/api/v1/seed/ranges/${lease!.id}/batches`, batch)
+            headers += batch.headers
+          },
+        })
+        await api.post(`/api/v1/seed/ranges/${lease.id}/done`)
+        const seconds = (Date.now() - started) / 1000
+        log(`done: ${headers.toLocaleString('en')} headers in ${seconds.toFixed(0)} s (${Math.round(headers / seconds).toLocaleString('en')}/s)`)
+      } catch (error: any) {
+        if (error instanceof Unauthorized) throw error
+        if (error instanceof LeaseLost) {
+          log(`Range ${lease.id} passed to another node; dropping it.`)
+          continue
+        }
+        const reason = error instanceof Stopping ? 'node stopping' : error.message
+        if (!(error instanceof Stopping)) log(`Range ${lease.id} failed: ${reason}`)
+        await api.post(`/api/v1/seed/ranges/${lease.id}/release`, { error: reason }).catch(() => {})
+        if (!this.stopping) await this.pause(30_000)
+      } finally {
+        client?.close()
+      }
+    }
+  }
+
+  /** Sleeps, but wakes up early when the node is stopping. */
+  private async pause(ms: number) {
+    for (let waited = 0; waited < ms && !this.stopping; waited += 1000) await sleep(1000)
+  }
+}
+
+class Unauthorized extends Error {}
+
+/** The seed API client: JSON over HTTPS, gzip request bodies, retries for network errors and 5xx. */
+class SeedApi {
+  constructor(
+    private base: string,
+    private token: string,
+    private warn: (message: string) => void
+  ) {}
+
+  async post(path: string, body?: unknown): Promise<unknown> {
+    const payload = body === undefined ? undefined : gzipSync(JSON.stringify(body))
+    for (let attempt = 1; ; attempt++) {
+      let status = 0
+      let text = ''
+      try {
+        const response = await fetch(this.base + path, {
+          method: 'POST',
+          headers: {
+            'authorization': `Bearer ${this.token}`,
+            'accept': 'application/json',
+            ...(payload ? { 'content-type': 'application/json', 'content-encoding': 'gzip' } : {}),
+          },
+          body: payload,
+          signal: AbortSignal.timeout(10 * 60 * 1000),
+        })
+        status = response.status
+        text = await response.text()
+      } catch (error: any) {
+        text = error.message
+      }
+      if (status === 204) return null
+      if (status >= 200 && status < 300) return text ? JSON.parse(text) : null
+      if (status === 401) throw new Unauthorized('The seed API rejected this node token')
+      if (status === 409) throw new LeaseLost(text)
+      if (status >= 400 && status < 500) throw new Error(`Seed API ${path}: ${status} ${text.slice(0, 300)}`)
+      // Network errors and 5xx: storing a batch twice is harmless, so try again.
+      if (attempt >= 8) throw new Error(`Seed API ${path}: ${status || 'no response'} ${text.slice(0, 300)}`)
+      const wait = Math.min(120_000, 2000 * 2 ** (attempt - 1))
+      this.warn(`Seed API ${path}: ${status || text}; retrying in ${wait / 1000} s`)
+      await sleep(wait)
+    }
+  }
+}

commands/SeedPlan.ts

@@ -0,0 +1,41 @@
+import { BaseCommand, args, flags } from '@adonisjs/core/ace'
+import { CommandOptions } from '@adonisjs/core/types/ace'
+import NntpService from '#services/NntpService'
+import SeedOrchestrator from '#services/SeedOrchestrator'
+
+/**
+ * Plans seed ranges (CONCEPT.md D38) for groups, from below what live
+ * indexing, the backfill and earlier plans cover down to `--days` back (or
+ * the server's oldest article). A group's backfill stops where its plan
+ * starts. Ranges are interleaved across groups, newest first.
+ *
+ *   node ace seed:plan alt.binaries.ath alt.binaries.mom --days 365
+ */
+export default class SeedPlan extends BaseCommand {
+  public static commandName = 'seed:plan'
+  public static description = 'Plans article ranges of groups for seeding nodes to lease.'
+
+  public static options: CommandOptions = {
+    startApp: true,
+  }
+
+  @args.spread({ description: 'Groups' })
+  declare groups: string[]
+
+  @flags.number({ description: 'How many days back (default: everything the server has)' })
+  declare days?: number
+
+  @flags.number({ description: 'Articles per range', default: 2_000_000 })
+  declare size: number
+
+  public async run() {
+    try {
+      const summary = await SeedOrchestrator.plan(this.groups, { days: this.days, size: this.size })
+      for (const s of summary) {
+        this.logger.info(`${s.group}: ${s.ranges.toLocaleString('en')} ranges, articles ${s.from.toLocaleString('en')}–${s.to.toLocaleString('en')}.`)
+      }
+    } finally {
+      await NntpService.shutdown()
+    }
+  }
+}

commands/SeedRange.ts

@@ -1,21 +1,19 @@
 import { BaseCommand, args, flags } from '@adonisjs/core/ace'
 import { CommandOptions } from '@adonisjs/core/types/ace'
 import nntpConfig from '#config/nntp'
-import { XoverClient, type Overview } from '#lib/nntp/xover_client'
-import IngestService, { segmentFromHeader, type IngestSegment } from '#services/IngestService'
-import StagingService from '#services/StagingService'
 import SeedService from '#services/SeedService'
 import QueueService from '#services/QueueService'
 import { isDeferred } from '#services/DeferredService'
-import { FileAggregator } from '#lib/aggregate/index'
+import { XoverClient } from '#lib/nntp/xover_client'
+import { runRange } from '#lib/seed/runner'
+import type { SeedBatch } from '#lib/seed/wire'
 
 /**
- * Edge aggregation (CONCEPT.md D37): indexes a range of a group's history
- * over one NNTP connection (gzip-compressed overviews when the server has
- * them), grouping headers into files in memory and
- * storing one row per file. Files cut by the range edges, or that never
- * complete, go through the regular ingest path as fragments. Header-only,
- * like the backfill: obfuscated subjects are counted, not downloaded.
+ * Edge aggregation (CONCEPT.md D37) on this machine: indexes a range of a
+ * group's history over one NNTP connection (gzip-compressed overviews when
+ * the server has them), grouping headers into files in memory and storing
+ * one row per file directly. seed:node does the same for a remote node,
+ * through the seed API. Header-only, like the backfill.
  *
  *   node ace seed:range alt.binaries.ath --start 46000000000 --count 2000000
  *   node ace seed:range alt.binaries.ath --start 46000000000 --count 2000000 --dry-run
@@ -49,17 +47,12 @@ export default class SeedRange extends BaseCommand {
   @flags.boolean({ description: 'Ask the server for gzip-compressed overviews', default: true })
   declare compress: boolean
 
-  /** Fragment batches being stored the regular way, off the node's critical path. */
-  private fragmentWork = new Set<Promise<void>>()
-  private fragmentError?: unknown
-
   public async run() {
     const first = Number(this.start)
-    const last = first + this.count - 1
     if (!Number.isSafeInteger(first) || first < 1) throw new Error(`--start must be an article number, got ${this.start}`)
-    const aggregator = new FileAggregator<IngestSegment>()
-    const stats = { headers: 0, segments: 0, obfuscated: 0, deferred: 0, complete: 0, files: 0, existing: 0, fragments: 0, inspections: 0 }
-    const time = { fetch: 0, parse: 0, store: 0 }
+    const last = first + this.count - 1
+    const stats = { headers: 0, obfuscated: 0, deferred: 0, complete: 0, files: 0, existing: 0, fragments: 0, inspections: 0 }
+    let storeMs = 0
     const cpuStart = process.cpuUsage()
     const started = Date.now()
 
@@ -74,56 +67,33 @@ export default class SeedRange extends BaseCommand {
     })
     this.logger.info(`Connected${client.compressed ? ', overviews gzip-compressed' : ''}.`)
     try {
-      await client.group(this.group)
-      const fetch = async (from: number) => {
-        const to = Math.min(from + this.chunk - 1, last)
-        const t = Date.now()
-        const overviews = await client.xover(from, to)
-        time.fetch += Date.now() - t
-        return overviews.filter(([n, h]) => Number.isFinite(n) && h['message-id'])
-      }
-
-      // One XOVER in flight while the previous batch is parsed and stored.
-      let from = first
-      const prefetch = (at: number) => {
-        const promise = fetch(at)
-        // Awaited after the current batch is stored; until then a failure must not count as unhandled.
-        promise.catch(() => {})
-        return promise
-      }
-      let next: Promise<Overview[]> | null = prefetch(from)
-      while (next) {
-        const overviews = await next
-        from += this.chunk
-        next = from <= last ? prefetch(from) : null
-
-        let t = Date.now()
-        const segments: IngestSegment[] = []
-        for (const [articleNumber, header] of overviews) {
-          const segment = segmentFromHeader(header, this.group, articleNumber)
-          if (segment) segments.push(segment)
-          else if (isDeferred(header)) stats.deferred++
-          else stats.obfuscated++
-        }
-        stats.headers += overviews.length
-        stats.segments += segments.length
-        const complete = aggregator.add(segments)
-        const fragments = aggregator.evict(this.idle)
-        stats.complete += complete.length
-        stats.fragments += fragments.length
-        time.parse += Date.now() - t
-
-        t = Date.now()
-        if (!this.dryRun) await this.store(complete, fragments, stats)
-        time.store += Date.now() - t
-        this.progress(stats, started)
-      }
-
-      const rest = aggregator.evict(0)
-      stats.fragments += rest.length
-      if (!this.dryRun) await this.store([], rest, stats)
-      await Promise.all(this.fragmentWork)
-      if (this.fragmentError) throw this.fragmentError
+      await runRange({
+        client,
+        group: this.group,
+        start: first,
+        end: last,
+        chunk: this.chunk,
+        idle: this.idle,
+        submitEvery: 1,
+        isDeferred,
+        sink: async (batch: SeedBatch) => {
+          stats.headers += batch.headers
+          stats.obfuscated += batch.obfuscated
+          stats.deferred += batch.deferred
+          stats.complete += batch.files.length
+          stats.fragments += batch.fragments.length
+          if (!this.dryRun) {
+            const t = Date.now()
+            const stored = await SeedService.store(batch.files)
+            await SeedService.storeFragments(this.group, batch.fragments)
+            storeMs += Date.now() - t
+            stats.files += stored.files
+            stats.existing += stored.existing
+            stats.inspections += stored.inspections
+          }
+          this.progress(stats, started)
+        },
+      })
     } finally {
       client.close()
       await QueueService.closeAll()
@@ -131,39 +101,14 @@ export default class SeedRange extends BaseCommand {
 
     const seconds = (Date.now() - started) / 1000
     const cpu = process.cpuUsage(cpuStart)
+    const n = (x: number) => x.toLocaleString('en')
     this.logger.info(
-      `${this.group} ${first}-${last}: ${stats.headers.toLocaleString('en')} headers in ${seconds.toFixed(0)} s ` +
-        `(${Math.round(stats.headers / seconds).toLocaleString('en')}/s). ` +
-        `${stats.segments.toLocaleString('en')} readable, ${stats.obfuscated.toLocaleString('en')} obfuscated, ${stats.deferred.toLocaleString('en')} deferred. ` +
-        `${stats.complete.toLocaleString('en')} complete files (${stats.files.toLocaleString('en')} new, ${stats.existing.toLocaleString('en')} known), ` +
-        `${stats.fragments.toLocaleString('en')} fragment segments, ${stats.inspections.toLocaleString('en')} queued for inspection.`
+      `${this.group} ${first}-${last}: ${n(stats.headers)} headers in ${seconds.toFixed(0)} s (${n(Math.round(stats.headers / seconds))}/s). ` +
+        `${n(stats.obfuscated)} obfuscated, ${n(stats.deferred)} deferred. ` +
+        `${n(stats.complete)} complete files (${n(stats.files)} new, ${n(stats.existing)} known), ` +
+        `${n(stats.fragments)} fragment segments, ${n(stats.inspections)} queued for inspection.`
     )
-    this.logger.info(
-      `Time: fetch ${(time.fetch / 1000).toFixed(0)} s (overlapped), parse+aggregate ${(time.parse / 1000).toFixed(0)} s, ` +
-        `store ${(time.store / 1000).toFixed(0)} s; node CPU ${((cpu.user + cpu.system) / 1e6).toFixed(0)} s.`
-    )
-  }
-
-  private async store(complete: Parameters<typeof SeedService.store>[0], fragments: IngestSegment[], stats: Record<string, number>) {
-    const stored = await SeedService.store(complete)
-    stats.files += stored.files
-    stats.existing += stored.existing
-    stats.inspections += stored.inspections
-    if (fragments.length === 0) return
-    // The regular path: staging (D35) for lone segments, binaries and parts for the rest. It runs
-    // beside fetching; with two batches in flight already, the node waits for one to finish.
-    while (this.fragmentWork.size >= 2) await Promise.race([...this.fragmentWork].map((w) => w.catch(() => {})))
-    if (this.fragmentError) throw this.fragmentError
-    const work = (async () => {
-      const { ingest } = StagingService.enabled(this.group)
-        ? await StagingService.filter(fragments, { counters: StagingService.countersEnabled(this.group) })
-        : { ingest: fragments }
-      await IngestService.ingest(ingest)
-    })()
-    this.fragmentWork.add(work)
-    work
-      .catch((error) => (this.fragmentError ??= error))
-      .finally(() => this.fragmentWork.delete(work))
+    this.logger.info(`Store ${(storeMs / 1000).toFixed(0)} s; node CPU ${((cpu.user + cpu.system) / 1e6).toFixed(0)} s.`)
   }
 
   private lastReport = 0

commands/SeedStatus.ts

@@ -0,0 +1,25 @@
+import { BaseCommand } from '@adonisjs/core/ace'
+import { CommandOptions } from '@adonisjs/core/types/ace'
+import SeedOrchestrator from '#services/SeedOrchestrator'
+
+/** Seeding progress per range status and per node (CONCEPT.md D38). */
+export default class SeedStatus extends BaseCommand {
+  public static commandName = 'seed:status'
+  public static description = 'Shows seeding progress per range status and per node.'
+
+  public static options: CommandOptions = {
+    startApp: true,
+  }
+
+  public async run() {
+    const status = await SeedOrchestrator.status()
+    const n = (x: unknown) => Number(x ?? 0).toLocaleString('en')
+    for (const r of status.ranges) {
+      this.logger.info(`${r.status}: ${n(r.ranges)} ranges, ${n(r.reached)} of ${n(r.articles)} articles reached, ${n(r.headers)} headers, ${n(r.files)} files`)
+    }
+    for (const node of status.nodes) {
+      this.logger.info(`node ${node.name}${node.disabled ? ' (disabled)' : ''}: ${node.leases} leases, ${n(node.headers)} headers, last seen ${node.last_seen_at ?? 'never'}`)
+    }
+    this.logger.info(`Headers in ranges finished in the last hour: ${n(status.headersDoneLastHour)}`)
+  }
+}

commands/SeedToken.ts

@@ -0,0 +1,37 @@
+import { BaseCommand, args, flags } from '@adonisjs/core/ace'
+import { CommandOptions } from '@adonisjs/core/types/ace'
+import db from '@adonisjs/lucid/services/db'
+import SeedOrchestrator from '#services/SeedOrchestrator'
+
+/**
+ * Creates a seeding node, or gives it a new token (the old one stops
+ * working), and prints the token once (CONCEPT.md D38).
+ *
+ *   node ace seed:token hetzner-1
+ *   node ace seed:token hetzner-1 --disable
+ */
+export default class SeedToken extends BaseCommand {
+  public static commandName = 'seed:token'
+  public static description = 'Creates a seeding node or replaces its token; prints the token once.'
+
+  public static options: CommandOptions = {
+    startApp: true,
+  }
+
+  @args.string({ description: 'Node name' })
+  declare name: string
+
+  @flags.boolean({ description: 'Disable the node instead (its leases pass to others after they expire)' })
+  declare disable: boolean
+
+  public async run() {
+    if (this.disable) {
+      const n = await db.from('seed_nodes').where('name', this.name).update({ disabled: true })
+      this.logger.info(n ? `${this.name}: disabled.` : `${this.name}: no such node.`)
+      return
+    }
+    const token = await SeedOrchestrator.issueToken(this.name)
+    this.logger.info(`${this.name}: token (shown once, keep it on the node only):`)
+    console.log(token)
+  }
+}