Close NNTP connections left idle for NNTP_IDLE_CLOSE_SECONDS
Test and deploy / test (push) Successful in 50s
Test and deploy / edge (push) Successful in 1m45s
Test and deploy / deploy (push) Successful in 2m20s

Eweka drops quiet connections, so the next command on one failed
("Spot inspection failed: NNTP connection closed"), and idle connections
held account slots that downloads could use. The coordinator pool now
closes free connections (and the plain one) on its sync once idle for
60 s; dry prober lanes, the spotter between rounds and the spot inspector
with nothing to inspect close theirs too, reconnecting when work comes.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
Daan MeijerandClaude Opus 5.5 committed 2026-10-22 14:31:49 +02:00
1 parent 705f9c3897
commit b60d716db0
8 files changed
+143 -4

No files matched your search

+7
View File
@@ -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 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 was paused and the coordinator's connections idle (LOG.md 2026-10-18
(75)). **Accepted** 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.
+39 -4
View File
@@ -12,7 +12,8 @@ const SHARE_MARGIN = 2
* have no NNTP account). At most NNTP_CONNECTIONS, reused; a connection * have no NNTP account). At most NNTP_CONNECTIONS, reused; a connection
* that failed, or is older than NNTP_CONNECTION_MAX_MINUTES, is closed and * 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 * 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 { class NntpPool {
private idle: XoverClient[] = [] private idle: XoverClient[] = []
@@ -22,6 +23,8 @@ class NntpPool {
private openedAt = new Map<XoverClient, number>() private openedAt = new Map<XoverClient, number>()
/** When a connection gets closed once free: its max age ±25%, so they don't all reconnect at once. */ /** When a connection gets closed once free: its max age ±25%, so they don't all reconnect at once. */
private retireAt = new Map<XoverClient, number>() private retireAt = new Map<XoverClient, number>()
/** When an idle connection came free. */
private idleSince = new Map<XoverClient, number>()
private limit = coordinatorConfig.nntpConnections private limit = coordinatorConfig.nntpConnections
/** Since the last sync: the most connections in use at once, and whether work waited for one (starved()). */ /** Since the last sync: the most connections in use at once, and whether work waited for one (starved()). */
private peak = 0 private peak = 0
@@ -50,6 +53,7 @@ class NntpPool {
this.limit = granted this.limit = granted
// Over the share: the idle ones close now, the busy ones once free (release()). // 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()!) while (this.open > this.limit && this.idle.length > 0) this.retire(this.idle.pop()!)
this.closeIdle()
// Under it with work waiting: connections for it. // Under it with work waiting: connections for it.
while (this.waiting.length > 0 && this.open < this.limit) { while (this.waiting.length > 0 && this.open < this.limit) {
const next = this.waiting.shift()! 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<XoverClient> { private async acquire(): Promise<XoverClient> {
// The one used last: the others stay free long enough to close.
const ready = this.idle.pop() const ready = this.idle.pop()
if (ready) return ready if (ready) {
this.idleSince.delete(ready)
return ready
}
if (this.open < this.size) { if (this.open < this.size) {
this.open++ this.open++
try { try {
@@ -134,13 +157,17 @@ class NntpPool {
} }
const next = this.waiting.shift() const next = this.waiting.shift()
if (next) next(client) if (next) next(client)
else this.idle.push(client) else {
this.idle.push(client)
this.idleSince.set(client, Date.now())
}
} }
private retire(client: XoverClient) { private retire(client: XoverClient) {
this.open-- this.open--
this.openedAt.delete(client) this.openedAt.delete(client)
this.retireAt.delete(client) this.retireAt.delete(client)
this.idleSince.delete(client)
try { try {
client.close() client.close()
} catch (error: any) { } catch (error: any) {
@@ -151,6 +178,9 @@ class NntpPool {
/** Closes every connection, also those in use: their commands fail. */ /** Closes every connection, also those in use: their commands fail. */
private plain: Promise<XoverClient> | null = null private plain: Promise<XoverClient> | null = null
private plainQueue: Promise<unknown> = Promise.resolve() private plainQueue: Promise<unknown> = 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, * 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. * cuts short (Leases.corrupt). Its replies are deflated here.
*/ */
async usePlain<T>(work: (client: XoverClient) => Promise<T>): Promise<T> { async usePlain<T>(work: (client: XoverClient) => Promise<T>): Promise<T> {
this.plainPending++
const run = this.plainQueue.then(async () => { const run = this.plainQueue.then(async () => {
this.plain ??= XoverClient.connect({ this.plain ??= XoverClient.connect({
host: env.get('NNTP_HOST')!, host: env.get('NNTP_HOST')!,
@@ -177,7 +208,10 @@ class NntpPool {
throw error throw error
} }
}) })
this.plainQueue = run.catch(() => {}) this.plainQueue = run.catch(() => {}).finally(() => {
this.plainPending--
this.plainUsedAt = Date.now()
})
return run return run
} }
@@ -188,6 +222,7 @@ class NntpPool {
this.busy.clear() this.busy.clear()
this.openedAt.clear() this.openedAt.clear()
this.retireAt.clear() this.retireAt.clear()
this.idleSince.clear()
this.open = 0 this.open = 0
} }
} }
+8
View File
@@ -121,6 +121,7 @@ export class Prober {
*/ */
private async lane() { private async lane() {
let client: XoverClient | null = null let client: XoverClient | null = null
let usedAt = Date.now()
try { try {
while (!this.stopped) { while (!this.stopped) {
if (this.running > this.share.granted) break if (this.running > this.share.granted) break
@@ -134,6 +135,12 @@ export class Prober {
this.stats.waitMs += Date.now() - waited this.stats.waitMs += Date.now() - waited
this.dry = batch.length === 0 this.dry = batch.length === 0
if (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 this.stats.idleMs += 30_000
await sleep(30_000) await sleep(30_000)
continue continue
@@ -170,6 +177,7 @@ export class Prober {
} }
} }
this.stats.inspectMs += Date.now() - inspected this.stats.inspectMs += Date.now() - inspected
usedAt = Date.now()
} catch (error: any) { } catch (error: any) {
logger.warn(`Probing ${batch[0].messageId} and ${batch.length - 1} more: ${error.message}`) logger.warn(`Probing ${batch[0].messageId} and ${batch.length - 1} more: ${error.message}`)
for (const candidate of batch) this.outcomes.push({ candidate, status: 'waiting' }) for (const candidate of batch) this.outcomes.push({ candidate, status: 'waiting' })
+6
View File
@@ -102,6 +102,12 @@ export class SpotInspector {
this.client = undefined this.client = undefined
} }
if (!round || round.inspected === 0) { 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)) for (let waited = 0; waited < IDLE_MS && !this.stopped; waited += 1000) await new Promise((r) => setTimeout(r, 1000))
} }
} }
+6
View File
@@ -60,6 +60,12 @@ export class Spotter {
this.client = undefined this.client = undefined
} }
const wait = coordinatorConfig.spots.everySeconds * 1000 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)) for (let waited = Date.now() - started; waited < wait && !this.stopped; waited += 1000) await new Promise((r) => setTimeout(r, 1000))
} }
} }
+7
View File
@@ -81,6 +81,13 @@ export default {
* under 2 minutes old as on ones over 10 (~25–35%; LOG.md 2026-10-09 (16)). * 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, 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: { probe: {
/** /**
* At most (its share, nntpShares): NNTP connections the prober reads articles with, besides the * At most (its share, nntpShares): NNTP connections the prober reads articles with, besides the
+1
View File
@@ -50,6 +50,7 @@ export default await Env.create(new URL('../', import.meta.url), {
NNTP_HEDGE_AFTER_SECONDS: Env.schema.number.optional(), NNTP_HEDGE_AFTER_SECONDS: Env.schema.number.optional(),
NNTP_HEDGE_CONNECTIONS: Env.schema.number.optional(), NNTP_HEDGE_CONNECTIONS: Env.schema.number.optional(),
NNTP_CONNECTION_MAX_MINUTES: 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(), NNTP_STALL_KB_PER_MINUTE: Env.schema.number.optional(),
// Newznab API key (empty: no key needed) // Newznab API key (empty: no key needed)
+69
View File
@@ -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)
})
})