usenet-indexer/commands/SeedNode.ts

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}`)
    // Logged in before leasing, so a node that can't reach its news server takes no range
    // (each lease counts as an attempt; a range is set aside after five). Kept across ranges.
    let client: XoverClient | undefined
    let idleSince = Date.now()
    try {
      while (!this.stopping) {
        // A news server may drop a connection that sat idle (nothing to lease): log in afresh.
        if (client && Date.now() - idleSince > 2 * 60_000) {
          client.close()
          client = undefined
        }
        if (!client) {
          try {
            client = await this.connect()
          } catch (error: any) {
            const ipLimit = /simultaneous IP/i.test(error.message)
            log(`NNTP login failed: ${error.message}; trying again in ${ipLimit ? 5 : 1} min`)
            await this.pause(ipLimit ? 300_000 : 60_000)
            continue
          }
        }
        const outcome = await this.range(api, client, log)
        if (outcome === 'replace') {
          client.close()
          client = undefined
        }
        if (outcome !== 'idle') idleSince = Date.now()
      }
    } finally {
      client?.close()
    }
  }

  private connect() {
    return XoverClient.connect({
      host: nntpConfig.host,
      port: nntpConfig.port,
      secure: nntpConfig.secure,
      servername: nntpConfig.tlsServername,
      user: nntpConfig.user,
      password: nntpConfig.password?.release(),
      timeoutMs: nntpConfig.commandTimeoutMs,
      compress: true,
    })
  }

  /**
   * Leases and works one range: 'done', 'idle' (nothing leased; the connection
   * wasn't used), or 'replace' (the connection may be in an odd state).
   */
  private async range(
    api: SeedApi,
    client: XoverClient,
    log: (message: string) => void
  ): Promise<'done' | 'idle' | 'replace'> {
    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)
      return 'idle'
    }
    if (!lease) {
      if (this.once) this.stopping = true
      else await this.pause(60_000)
      return 'idle'
    }

    log(`${lease.group} ${lease.start}–${lease.end}`)
    const started = Date.now()
    let headers = 0
    let chain: Promise<void> = Promise.resolve()
    const posting = new Set<Promise<void>>()
    // Throws once a post has failed, so the range stops fetching.
    const failedSoFar = () => Promise.race([chain, Promise.resolve()])
    try {
      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'] ?? '')),
        // Batches are posted in order, one after the other, while fetching goes on; with two
        // waiting, fetching waits too. A failed post stops the chain, so no batch is stored
        // past a missing one (the range would resume beyond it).
        sink: async (batch: SeedBatch) => {
          if (this.stopping) throw new Stopping()
          while (posting.size >= 2) await Promise.race(posting)
          const post = chain.then(async () => {
            await api.post(`/api/v1/seed/ranges/${lease!.id}/batches`, batch)
            headers += batch.headers
          })
          chain = post
          posting.add(post)
          post.catch(() => {}).finally(() => posting.delete(post))
          await failedSoFar()
        },
      })
      await chain
      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)`
      )
      return 'done'
    } catch (error: any) {
      if (error instanceof Unauthorized) throw error
      // An XOVER may still be in flight: the connection is replaced either way.
      if (error instanceof LeaseLost) {
        log(`Range ${lease.id} passed to another node; dropping it.`)
        return 'replace'
      }
      const reason = error instanceof Stopping ? 'node stopping' : error.message
      if (!(error instanceof Stopping)) log(`Range ${lease.id} failed: ${reason}`)
      // Let posts in flight finish (or fail) first, so the range resumes after what was stored.
      await chain.catch(() => {})
      await api.post(`/api/v1/seed/ranges/${lease.id}/release`, { error: reason }).catch(() => {})
      if (!this.stopping) await this.pause(30_000)
      return 'replace'
    }
  }

  /** 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)
      // 404 comes from the proxy while the seed API restarts (a deploy): this API has no 404s of its own.
      if (status >= 400 && status < 500 && status !== 404)
        throw new Error(`Seed API ${path}: ${status} ${text.slice(0, 300)}`)
      // Network errors, 404 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)
    }
  }
}