Daan MeijerandClaude Opus 5.5
883d85e91d
It waited for every chunk in flight, and a connection the server held back (after killed coordinators left sessions open) waits up to the 10-minute reply timeout: it never seemed to stop. It now waits at most 15 s, then closes all connections, busy ones too; the greeting and login time out after 30 s. And run() ends with terminate(), as edge:node does. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
98 lines
4.1 KiB
TypeScript
98 lines
4.1 KiB
TypeScript
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 Settler from '#services/settler'
|
|
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
|
|
const STOP_WAIT_MS = 15 * 1000
|
|
|
|
/**
|
|
* The coordinator's background work (REBUILD.md, part 1): the prefetcher,
|
|
* merging fragments of finished ranges every minute, and every hour
|
|
* planning ranges for new articles and settling quiet releases. 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<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('Coordinator started.')
|
|
// Two loops: the prefetcher's ticks don't wait for merging or planning (a merge round
|
|
// of up to MERGE_BUDGET_MS 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)
|
|
}
|
|
})()
|
|
let plannedAt = 0
|
|
let mergedAt = 0
|
|
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.`)
|
|
const settled = await Settler.run()
|
|
if (settled > 0) this.logger.info(`Settled ${settled} releases.`)
|
|
}
|
|
if (Date.now() - mergedAt >= MERGE_EVERY_MS) {
|
|
mergedAt = Date.now()
|
|
// Rounds until a whole pass over the file keys decides nothing more, or MERGE_BUDGET_MS.
|
|
// (Going on while only waiting files were left kept Postgres busy all the time.)
|
|
const totals = { completed: 0, incomplete: 0, waiting: 0 }
|
|
let decidedThisPass = 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
|
|
decidedThisPass += merged.completed + merged.incomplete
|
|
if (merged.wrapped) {
|
|
if (decidedThisPass === 0) break
|
|
decidedThisPass = 0
|
|
}
|
|
}
|
|
if (totals.completed || totals.incomplete) this.logger.info(`Fragments: ${totals.completed} files completed, ${totals.incomplete} incomplete, ${totals.waiting} waiting.`)
|
|
}
|
|
} catch (error: any) {
|
|
this.logger.error(`Coordinator round failed: ${error.message}`)
|
|
}
|
|
await sleep(TICK_MS)
|
|
}
|
|
await prefetching
|
|
// 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()
|
|
this.logger.info('Coordinator stopped.')
|
|
stopped!()
|
|
await this.terminate()
|
|
}
|
|
}
|