From 883d85e91d4b79d59d7c358cb0d361cb289c94d1 Mon Sep 17 00:00:00 2001 From: Daan Meijer Date: Fri, 9 Oct 2026 03:20:11 +0200 Subject: [PATCH] Rebuild: the coordinator stops on SIGTERM 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 --- app/services/nntp_pool.ts | 7 ++++++- commands/coordinator_run.ts | 9 ++++++++- lib/nntp/xover_client.ts | 3 ++- 3 files changed, 16 insertions(+), 3 deletions(-) diff --git a/app/services/nntp_pool.ts b/app/services/nntp_pool.ts index d21927f..8aef5a4 100644 --- a/app/services/nntp_pool.ts +++ b/app/services/nntp_pool.ts @@ -10,6 +10,7 @@ import { XoverClient } from '#lib/nntp/index' */ class NntpPool { private idle: XoverClient[] = [] + private busy = new Set() private open = 0 private waiting: ((client: XoverClient) => void)[] = [] @@ -19,6 +20,7 @@ class NntpPool { async use(work: (client: XoverClient) => Promise): Promise { const client = await this.acquire() + this.busy.add(client) let broken = false try { return await work(client) @@ -26,6 +28,7 @@ class NntpPool { broken = true throw error } finally { + this.busy.delete(client) this.release(client, broken) } } @@ -71,8 +74,10 @@ class NntpPool { else this.idle.push(client) } + /** Closes every connection, also those in use: their commands fail. */ async close() { - for (const client of this.idle.splice(0)) client.close() + for (const client of [...this.idle.splice(0), ...this.busy]) client.close() + this.busy.clear() this.open = 0 } } diff --git a/commands/coordinator_run.ts b/commands/coordinator_run.ts index 6e06e3c..325ceb4 100644 --- a/commands/coordinator_run.ts +++ b/commands/coordinator_run.ts @@ -11,6 +11,7 @@ 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, @@ -30,6 +31,7 @@ export default class CoordinatorRun extends BaseCommand { 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 () => { + this.logger.info('Stopping: waiting for the chunks in flight.') stopping = true prefetcher.stop() await done @@ -83,8 +85,13 @@ export default class CoordinatorRun extends BaseCommand { await sleep(TICK_MS) } await prefetching - while (prefetcher.active > 0) await sleep(200) + // 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() } } diff --git a/lib/nntp/xover_client.ts b/lib/nntp/xover_client.ts index 0e7dac4..b004b89 100644 --- a/lib/nntp/xover_client.ts +++ b/lib/nntp/xover_client.ts @@ -67,7 +67,8 @@ export class XoverClient { ? tls.connect({ host: options.host, port: options.port, servername: options.servername || options.host }) : net.connect({ host: options.host, port: options.port }) const client = new XoverClient(sock, options.timeoutMs ?? 120_000) - sock.setTimeout(client.timeoutMs) + // The greeting and login are short: a server holding a connection back (too many open) fails it in 30 s, not timeoutMs. + sock.setTimeout(Math.min(client.timeoutMs, 30_000)) try { await client.expect(await client.line(), ['200', '201']) if (options.user) {