diff --git a/REBUILD.md b/REBUILD.md index e3c1a0f..570fc8a 100644 --- a/REBUILD.md +++ b/REBUILD.md @@ -344,3 +344,10 @@ g. The account's NNTP connections shared by priority (`nntp_shares`): the fixed split (38 + 8), under which the prober stayed at 8 while history was paused and the coordinator's connections idle (LOG.md 2026-10-18 (75)). **Accepted** 2026-10-18. +h. Idle NNTP connections closed: one with nothing to do for + `NNTP_IDLE_CLOSE_SECONDS` (60) is closed and opened again with work + (the coordinator's pool on its next sync, the plain one too; a dry + prober lane; the spotter between rounds; the spot inspector with no + spots). Eweka dropped the quiet ones and the next command failed + ("Spot inspection failed: NNTP connection closed"), and idle ones took + the account's connections from downloads. **Accepted** 2026-10-22. diff --git a/app/services/nntp_pool.ts b/app/services/nntp_pool.ts index a7d7e98..b5523f5 100644 --- a/app/services/nntp_pool.ts +++ b/app/services/nntp_pool.ts @@ -12,7 +12,8 @@ const SHARE_MARGIN = 2 * have no NNTP account). At most NNTP_CONNECTIONS, reused; a connection * that failed, or is older than NNTP_CONNECTION_MAX_MINUTES, is closed and * replaced on the next use. Sharing the account (share()), at most its - * share: over it, connections close as they come free. + * share: over it, connections close as they come free. One left free for + * NNTP_IDLE_CLOSE_SECONDS is closed on the next sync. */ class NntpPool { private idle: XoverClient[] = [] @@ -22,6 +23,8 @@ class NntpPool { private openedAt = new Map() /** When a connection gets closed once free: its max age ±25%, so they don't all reconnect at once. */ private retireAt = new Map() + /** When an idle connection came free. */ + private idleSince = new Map() private limit = coordinatorConfig.nntpConnections /** Since the last sync: the most connections in use at once, and whether work waited for one (starved()). */ private peak = 0 @@ -50,6 +53,7 @@ class NntpPool { this.limit = granted // Over the share: the idle ones close now, the busy ones once free (release()). while (this.open > this.limit && this.idle.length > 0) this.retire(this.idle.pop()!) + this.closeIdle() // Under it with work waiting: connections for it. while (this.waiting.length > 0 && this.open < this.limit) { const next = this.waiting.shift()! @@ -93,9 +97,28 @@ class NntpPool { } } + /** Closes the connections free for nntpIdleCloseMs, the plain one too. */ + private closeIdle() { + const idleMs = coordinatorConfig.nntpIdleCloseMs + if (idleMs <= 0) return + const before = Date.now() - idleMs + for (const client of this.idle.filter((c) => (this.idleSince.get(c) ?? Infinity) <= before)) { + this.idle.splice(this.idle.indexOf(client), 1) + this.retire(client) + } + if (this.plain && this.plainPending === 0 && this.plainUsedAt <= before) { + void this.plain.then((c) => c.close()).catch(() => {}) + this.plain = null + } + } + private async acquire(): Promise { + // The one used last: the others stay free long enough to close. const ready = this.idle.pop() - if (ready) return ready + if (ready) { + this.idleSince.delete(ready) + return ready + } if (this.open < this.size) { this.open++ try { @@ -134,13 +157,17 @@ class NntpPool { } const next = this.waiting.shift() if (next) next(client) - else this.idle.push(client) + else { + this.idle.push(client) + this.idleSince.set(client, Date.now()) + } } private retire(client: XoverClient) { this.open-- this.openedAt.delete(client) this.retireAt.delete(client) + this.idleSince.delete(client) try { client.close() } catch (error: any) { @@ -151,6 +178,9 @@ class NntpPool { /** Closes every connection, also those in use: their commands fail. */ private plain: Promise | null = null private plainQueue: Promise = Promise.resolve() + /** Commands queued or running on the plain connection, and when the last one ended. */ + private plainPending = 0 + private plainUsedAt = 0 /** * One connection without XFEATURE COMPRESS GZIP, opened when first needed, @@ -158,6 +188,7 @@ class NntpPool { * cuts short (Leases.corrupt). Its replies are deflated here. */ async usePlain(work: (client: XoverClient) => Promise): Promise { + this.plainPending++ const run = this.plainQueue.then(async () => { this.plain ??= XoverClient.connect({ host: env.get('NNTP_HOST')!, @@ -177,7 +208,10 @@ class NntpPool { throw error } }) - this.plainQueue = run.catch(() => {}) + this.plainQueue = run.catch(() => {}).finally(() => { + this.plainPending-- + this.plainUsedAt = Date.now() + }) return run } @@ -188,6 +222,7 @@ class NntpPool { this.busy.clear() this.openedAt.clear() this.retireAt.clear() + this.idleSince.clear() this.open = 0 } } diff --git a/app/services/prober.ts b/app/services/prober.ts index 3ec7ea3..e6620e9 100644 --- a/app/services/prober.ts +++ b/app/services/prober.ts @@ -121,6 +121,7 @@ export class Prober { */ private async lane() { let client: XoverClient | null = null + let usedAt = Date.now() try { while (!this.stopped) { if (this.running > this.share.granted) break @@ -134,6 +135,12 @@ export class Prober { this.stats.waitMs += Date.now() - waited this.dry = batch.length === 0 if (batch.length === 0) { + // Nothing to probe for a while: the connection closes (nntpIdleCloseMs), and opens again with work. + const idleMs = coordinatorConfig.nntpIdleCloseMs + if (client && idleMs > 0 && Date.now() - usedAt >= idleMs) { + client.close() + client = null + } this.stats.idleMs += 30_000 await sleep(30_000) continue @@ -170,6 +177,7 @@ export class Prober { } } this.stats.inspectMs += Date.now() - inspected + usedAt = Date.now() } catch (error: any) { logger.warn(`Probing ${batch[0].messageId} and ${batch.length - 1} more: ${error.message}`) for (const candidate of batch) this.outcomes.push({ candidate, status: 'waiting' }) diff --git a/app/services/spot_inspector.ts b/app/services/spot_inspector.ts index 9ed5a83..051528b 100644 --- a/app/services/spot_inspector.ts +++ b/app/services/spot_inspector.ts @@ -102,6 +102,12 @@ export class SpotInspector { this.client = undefined } if (!round || round.inspected === 0) { + // Nothing to inspect for a while: closed (nntpIdleCloseMs), opened again with work. + const idleMs = coordinatorConfig.nntpIdleCloseMs + if (idleMs > 0 && IDLE_MS >= idleMs) { + this.client?.close() + this.client = undefined + } for (let waited = 0; waited < IDLE_MS && !this.stopped; waited += 1000) await new Promise((r) => setTimeout(r, 1000)) } } diff --git a/app/services/spotter.ts b/app/services/spotter.ts index 0520a36..823d67c 100644 --- a/app/services/spotter.ts +++ b/app/services/spotter.ts @@ -60,6 +60,12 @@ export class Spotter { this.client = undefined } const wait = coordinatorConfig.spots.everySeconds * 1000 + // Quiet until the next round: closed (nntpIdleCloseMs), opened again then. + const idleMs = coordinatorConfig.nntpIdleCloseMs + if (idleMs > 0 && wait - (Date.now() - started) >= idleMs) { + this.client?.close() + this.client = undefined + } for (let waited = Date.now() - started; waited < wait && !this.stopped; waited += 1000) await new Promise((r) => setTimeout(r, 1000)) } } diff --git a/config/coordinator.ts b/config/coordinator.ts index 3b729c8..3717ccd 100644 --- a/config/coordinator.ts +++ b/config/coordinator.ts @@ -81,6 +81,13 @@ export default { * under 2 minutes old as on ones over 10 (~25–35%; LOG.md 2026-10-09 (16)). */ nntpConnectionMaxMs: env.get('NNTP_CONNECTION_MAX_MINUTES', 0) * 60 * 1000, + /** + * A connection with nothing to do for this long is closed, and opened again + * when there's work. 0: kept. Eweka drops connections that stay quiet, and + * the next command on one failed ("Spot inspection failed: NNTP connection + * closed"); idle ones also take the account's connections from downloads. + */ + nntpIdleCloseMs: env.get('NNTP_IDLE_CLOSE_SECONDS', 60) * 1000, probe: { /** * At most (its share, nntpShares): NNTP connections the prober reads articles with, besides the diff --git a/start/env.ts b/start/env.ts index 3c49297..6db9cf2 100644 --- a/start/env.ts +++ b/start/env.ts @@ -50,6 +50,7 @@ export default await Env.create(new URL('../', import.meta.url), { NNTP_HEDGE_AFTER_SECONDS: Env.schema.number.optional(), NNTP_HEDGE_CONNECTIONS: Env.schema.number.optional(), NNTP_CONNECTION_MAX_MINUTES: Env.schema.number.optional(), + NNTP_IDLE_CLOSE_SECONDS: Env.schema.number.optional(), NNTP_STALL_KB_PER_MINUTE: Env.schema.number.optional(), // Newznab API key (empty: no key needed) diff --git a/tests/unit/nntp/pool.spec.ts b/tests/unit/nntp/pool.spec.ts new file mode 100644 index 0000000..6cbaca9 --- /dev/null +++ b/tests/unit/nntp/pool.spec.ts @@ -0,0 +1,69 @@ +import { test } from '@japa/runner' +import net from 'node:net' +import env from '#start/env' +import coordinatorConfig from '#config/coordinator' +import pool from '#services/nntp_pool' +import type { NntpShare } from '#services/nntp_shares' + +/** A tiny NNTP server that only logs in, counting the connections open. */ +function server(): Promise<{ port: number; open: () => number; close: () => void }> { + return new Promise((resolve) => { + let open = 0 + const srv = net.createServer((sock) => { + open++ + sock.on('close', () => open--) + sock.write('200 welcome\r\n') + sock.on('data', (chunk) => { + for (const line of chunk.toString().split('\r\n')) { + if (line.startsWith('AUTHINFO USER')) sock.write('381 more\r\n') + else if (line.startsWith('AUTHINFO PASS')) sock.write('281 ok\r\n') + else if (line === 'XFEATURE COMPRESS GZIP') sock.write('290 Feature Enabled\r\n') + else if (line === 'QUIT') sock.end('205 bye\r\n') + } + }) + }) + srv.listen(0, '127.0.0.1', () => + resolve({ + port: (srv.address() as net.AddressInfo).port, + open: () => open, + close: () => srv.close(), + }) + ) + }) +} + +const sleep = (ms: number) => new Promise((r) => setTimeout(r, ms)) +const share = { sync: async () => 4 } as unknown as NntpShare + +test.group('nntp / pool', () => { + test('a connection free for nntpIdleCloseMs closes on the next sync; one used since stays', async ({ + assert, + cleanup, + }) => { + const { port, open, close } = await server() + const idleCloseMs = coordinatorConfig.nntpIdleCloseMs + env.set('NNTP_HOST', '127.0.0.1') + env.set('NNTP_PORT', port) + env.set('NNTP_SECURE', false) + coordinatorConfig.nntpIdleCloseMs = 100 + cleanup(async () => { + coordinatorConfig.nntpIdleCloseMs = idleCloseMs + await pool.close() + close() + }) + await pool.share(share) + // Two at once: two connections. + await Promise.all([pool.use(() => sleep(20)), pool.use(() => sleep(20))]) + assert.equal(open(), 2) + await sleep(150) + // One used again: the other has been free for longer than 100 ms. + await pool.use(async () => {}) + await pool.share(share) + await sleep(50) + assert.equal(open(), 1) + await sleep(150) + await pool.share(share) + await sleep(50) + assert.equal(open(), 0) + }) +})