usenet-indexer/commands/SeedRange.ts

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'

/**
 * 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.
 *
 *   node ace seed:range alt.binaries.ath --start 46000000000 --count 2000000
 *   node ace seed:range alt.binaries.ath --start 46000000000 --count 2000000 --dry-run
 */
export default class SeedRange extends BaseCommand {
  public static commandName = 'seed:range'
  public static description = "Indexes a range of a group's headers by aggregating them into files on this node."

  public static options: CommandOptions = {
    startApp: true,
  }

  @args.string({ description: 'Group' })
  declare group: string

  @flags.string({ description: 'First article number' })
  declare start: string

  @flags.number({ description: 'Articles to index', default: 1_000_000 })
  declare count: number

  @flags.number({ description: 'Articles per XOVER', default: 20_000 })
  declare chunk: number

  @flags.number({ description: 'Batches without a new part before an open file is given up as fragments', default: 5 })
  declare idle: number

  @flags.boolean({ description: 'Fetch and aggregate only; store nothing' })
  declare dryRun: boolean

  @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 cpuStart = process.cpuUsage()
    const started = Date.now()

    const client = await XoverClient.connect({
      host: nntpConfig.host,
      port: nntpConfig.port,
      secure: nntpConfig.secure,
      user: nntpConfig.user,
      password: nntpConfig.password?.release(),
      timeoutMs: nntpConfig.commandTimeoutMs,
      compress: this.compress,
    })
    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
    } finally {
      client.close()
      await QueueService.closeAll()
    }

    const seconds = (Date.now() - started) / 1000
    const cpu = process.cpuUsage(cpuStart)
    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.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))
  }

  private lastReport = 0
  private progress(stats: Record<string, number>, started: number) {
    if (Date.now() - this.lastReport < 30_000) return
    this.lastReport = Date.now()
    const seconds = (Date.now() - started) / 1000
    this.logger.info(
      `${stats.headers.toLocaleString('en')} headers (${Math.round(stats.headers / seconds).toLocaleString('en')}/s), ` +
        `${stats.complete.toLocaleString('en')} files, ${stats.fragments.toLocaleString('en')} fragment segments.`
    )
  }
}