usenet-indexer/commands/coordinator_run.ts

import { BaseCommand, flags } from '@adonisjs/core/ace'
import type { CommandOptions } from '@adonisjs/core/types/ace'
import { setTimeout as sleep } from 'node:timers/promises'
import { Prefetcher } from '#services/prefetcher'
import Planner from '#services/planner'
import Leases from '#services/leases'
import Settler from '#services/settler'
import NntpPool from '#services/nntp_pool'
import { NntpShare } from '#services/nntp_shares'
import coordinatorConfig from '#config/coordinator'

const TICK_MS = 2000
const PLAN_EVERY_MS = 60 * 60 * 1000
const STOP_WAIT_MS = 15 * 1000
/** Leases whose worker went quiet after reading everything are finished every few minutes. */
const ABANDONED_EVERY_MS = 5 * 60 * 1000

/**
 * The coordinator's background work (REBUILD.md, part 1): the prefetcher, the tail's ranges every few minutes,
 * new groups' history every hour, and abandoned leases. The API (leases, stream, batches) runs in the web server.
 * Voyager 3: the settler and the fragment merger are gone (the window finishes a release once; the block merger
 * joins edge releases).
 */
export default class CoordinatorRun extends BaseCommand {
  static commandName = 'coordinator:run'
  static description = 'Prefetches ranges into the spool and plans new ones'
  static options: CommandOptions = { startApp: true, staysAlive: true }

  @flags.string({
    description:
      'This process\'s share of the prefetching, "i/n" (ranges with id % n = i - 1); only 1/n plans, merges and settles',
  })
  declare shard?: string

  async run() {
    const [index, count] = (this.shard ?? '1/1').split('/').map(Number)
    if (!(count >= 1 && index >= 1 && index <= count))
      throw new Error(`--shard: "i/n" with 1 <= i <= n, not "${this.shard}"`)
    const prefetcher = new Prefetcher({ index: index - 1, count })
    const share = new NntpShare(
      count > 1 ? `coordinator ${index}/${count}` : 'coordinator',
      coordinatorConfig.nntpShares.coordinatorPriority,
      coordinatorConfig.nntpMinConnections,
      coordinatorConfig.nntpConnections
    )
    const main = index === 1
    let stopping = false
    let stopped: () => void
    const done = new Promise<void>((resolve) => (stopped = resolve))
    // bin/console turns SIGTERM into app.terminate: wait for the spooling in flight first.
    this.app.terminating(async () => {
      this.logger.info('Stopping: waiting for the chunks in flight.')
      stopping = true
      prefetcher.stop()
      await done
    })
    this.logger.info(
      count > 1
        ? `Coordinator started (shard ${index}/${count}${main ? '' : ', prefetching only'}).`
        : 'Coordinator started.'
    )
    // The pool's size: its share of the account's connections, the prober the rest (app/services/nntp_shares.ts).
    // Before the first connection: until then it's NNTP_CONNECTIONS.
    const first = await NntpPool.share(share)
    this.logger.info(`NNTP share: ${first.granted} connections.`)
    const sharing = (async () => {
      while (!stopping) {
        try {
          const { was, granted } = await NntpPool.share(share)
          if (granted !== was) this.logger.info(`NNTP share: ${granted} connections (was ${was}).`)
        } catch (error: any) {
          this.logger.error(`NNTP share failed: ${error.message}`)
        }
        for (
          let waited = 0;
          waited < coordinatorConfig.nntpShares.everyMs && !stopping;
          waited += TICK_MS
        )
          await sleep(TICK_MS)
      }
    })()
    // The prefetcher's ticks in a loop of their own: planning doesn't hold them (a merge round of up to 30 s once
    // left the connections idle once their targets were fetched).
    const prefetching = (async () => {
      while (!stopping) {
        try {
          await prefetcher.tick()
        } catch (error: any) {
          this.logger.error(`Prefetch round failed: ${error.message}`)
        }
        await sleep(TICK_MS)
      }
    })()
    const abandoned = (async () => {
      while (main && !stopping) {
        try {
          const finished = await Leases.finishAbandoned()
          if (finished > 0)
            this.logger.info(
              `Finished ${finished} ranges read to their end whose worker went quiet.`
            )
        } catch (error: any) {
          this.logger.error(`Finishing abandoned leases failed: ${error.message}`)
        }
        try {
          const settled = await Settler.settleComplete()
          if (settled > 0)
            this.logger.info(`Settled ${settled} complete provisional releases quiet for hours.`)
        } catch (error: any) {
          this.logger.error(`Settling complete releases failed: ${error.message}`)
        }
        for (let waited = 0; waited < ABANDONED_EVERY_MS && !stopping; waited += TICK_MS)
          await sleep(TICK_MS)
      }
    })()
    // History for new groups hourly; the tail every few minutes.
    let plannedAt = 0
    let tailedAt = 0
    while (!stopping) {
      try {
        if (main && Date.now() - plannedAt >= PLAN_EVERY_MS) {
          plannedAt = Date.now()
          const { groups, ranges } = await Planner.planAll()
          if (ranges > 0) this.logger.info(`Planned ${ranges} ranges over ${groups} groups.`)
        }
        if (main && Date.now() - tailedAt >= coordinatorConfig.tail.everySeconds * 1000) {
          tailedAt = Date.now()
          const { groups, ranges, extended } = await Planner.planTail()
          if (ranges + extended > 0)
            this.logger.info(
              `Tail: ${ranges} ranges added, ${extended} grown, for ${groups} groups.`
            )
        }
      } catch (error: any) {
        this.logger.error(`Coordinator round failed: ${error.message}`)
      }
      await sleep(TICK_MS)
    }
    await Promise.all([prefetching, abandoned, sharing])
    // The chunks in flight, for at most STOP_WAIT_MS (a connection the server holds back can take minutes); then closed.
    const waitUntil = Date.now() + STOP_WAIT_MS
    while (prefetcher.active > 0 && Date.now() < waitUntil) await sleep(200)
    if (prefetcher.active > 0)
      this.logger.info(`Closing ${prefetcher.active} connections still fetching.`)
    await NntpPool.close()
    await share.leave().catch(() => {})
    this.logger.info('Coordinator stopped.')
    stopped!()
    await this.terminate()
  }
}