import { BaseCommand } 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 FragmentMerger from '#services/fragment_merger' import NntpPool from '#services/nntp_pool' const TICK_MS = 2000 const PLAN_EVERY_MS = 60 * 60 * 1000 const MERGE_EVERY_MS = 60 * 1000 const MERGE_BUDGET_MS = 30 * 1000 /** * The coordinator's background work (REBUILD.md, part 1): the prefetcher, * merging fragments of finished ranges every minute, and planning ranges * for new articles every hour. The API (leases, stream, * batches) runs in the web server. */ 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 } async run() { const prefetcher = new Prefetcher() let stopping = false let stopped: () => void const done = new Promise((resolve) => (stopped = resolve)) // bin/console turns SIGTERM into app.terminate: wait for the spooling in flight first. this.app.terminating(async () => { stopping = true prefetcher.stop() await done }) let plannedAt = 0 let mergedAt = 0 this.logger.info('Coordinator started.') while (!stopping) { try { if (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 (Date.now() - mergedAt >= MERGE_EVERY_MS) { mergedAt = Date.now() // Rounds until nothing more can be decided, or MERGE_BUDGET_MS. const totals = { completed: 0, incomplete: 0, waiting: 0 } while (!stopping && Date.now() - mergedAt < MERGE_BUDGET_MS) { const merged = await FragmentMerger.run() totals.completed += merged.completed totals.incomplete += merged.incomplete totals.waiting = merged.waiting if (merged.files === 0) break } if (totals.completed || totals.incomplete) this.logger.info(`Fragments: ${totals.completed} files completed, ${totals.incomplete} incomplete, ${totals.waiting} waiting.`) } await prefetcher.tick() } catch (error: any) { this.logger.error(`Coordinator round failed: ${error.message}`) } await sleep(TICK_MS) } while (prefetcher.active > 0) await sleep(200) await NntpPool.close() stopped!() } }