Seed relay: dead NNTP connections found in 90 s and retried
Five ranges "never answered" in 10 minutes, yet their chunk came back in 0.3 s on a fresh connection: the news server drops idle connections without telling, and the pool's next request on one waited for the 10-minute inactivity timeout. Now: SEED_NNTP_TIMEOUT_MS 90 s (inactivity; slow ranges keep streaming), one retry on a fresh connection, and pooled connections idle over 60 s are replaced before use. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
1 parent
8d86f7d489
commit
028dd90a6f
4 files changed
+72
-5
No files matched your search
@@ -8,8 +8,12 @@ interface Slot {
|
||||
usedAt: number
|
||||
}
|
||||
|
||||
/** A pooled connection unused this long is replaced: the news server may have dropped it. */
|
||||
const IDLE_MS = 2 * 60 * 1000
|
||||
/**
|
||||
* A pooled connection unused this long is replaced: the news server drops
|
||||
* idle connections without telling, and a request on one then hangs until
|
||||
* the inactivity timeout (LOG.md 2026-10-05 (11)).
|
||||
*/
|
||||
const IDLE_MS = 60 * 1000
|
||||
|
||||
/**
|
||||
* NNTP connections in the seed API for relay nodes (CONCEPT.md D38): the
|
||||
@@ -26,7 +30,16 @@ class SeedFetchPool {
|
||||
return indexerConfig.seed.nntpConnections
|
||||
}
|
||||
|
||||
/** Retried once on a fresh connection: a failure is nearly always a dead connection, not the articles. */
|
||||
async fetch(group: string, from: number, to: number): Promise<Buffer> {
|
||||
try {
|
||||
return await this.fetchOnce(group, from, to)
|
||||
} catch {
|
||||
return this.fetchOnce(group, from, to)
|
||||
}
|
||||
}
|
||||
|
||||
private async fetchOnce(group: string, from: number, to: number): Promise<Buffer> {
|
||||
const slot = await this.acquire()
|
||||
let ok = false
|
||||
try {
|
||||
@@ -42,6 +55,14 @@ class SeedFetchPool {
|
||||
}
|
||||
}
|
||||
|
||||
/** Closes the idle connections (tests, shutdown); busy ones close when released. */
|
||||
closeIdle() {
|
||||
for (const slot of this.idle.splice(0)) {
|
||||
slot.client.close()
|
||||
this.open--
|
||||
}
|
||||
}
|
||||
|
||||
private async acquire(): Promise<Slot> {
|
||||
while (true) {
|
||||
const slot = this.idle.pop()
|
||||
|
||||
+6
-2
@@ -66,8 +66,12 @@ export default {
|
||||
seed: {
|
||||
/** NNTP connections the seed API fetches with for relay nodes (CONCEPT.md D38); 0: relay off. */
|
||||
nntpConnections: env.get('SEED_NNTP_CONNECTIONS', 0),
|
||||
/** A relay XOVER may take this long: some old ranges come at ~1k headers/s (and slower when cold). */
|
||||
nntpTimeoutMs: env.get('SEED_NNTP_TIMEOUT_MS', 10 * 60 * 1000),
|
||||
/**
|
||||
* Silence that counts as a dead connection during a relay XOVER (an
|
||||
* inactivity timeout: slow ranges keep sending). Chunks that "never
|
||||
* answered" in 10 minutes came in 0.3 s on a fresh connection.
|
||||
*/
|
||||
nntpTimeoutMs: env.get('SEED_NNTP_TIMEOUT_MS', 90 * 1000),
|
||||
},
|
||||
|
||||
backfill: {
|
||||
|
||||
@@ -10,7 +10,7 @@ export interface XoverClientOptions {
|
||||
secure?: boolean
|
||||
/** Hostname the TLS certificate must match, when connecting through a tunnel (default: host). */
|
||||
servername?: string
|
||||
/** A command whose reply stalls this long fails, and the connection with it. */
|
||||
/** A reply with no data for this long fails, and the connection with it (inactivity, not duration). */
|
||||
timeoutMs?: number
|
||||
/** Ask for gzip-compressed overview data (XFEATURE COMPRESS GZIP), when the server has it. */
|
||||
compress?: boolean
|
||||
|
||||
@@ -0,0 +1,42 @@
|
||||
import { test } from '@japa/runner'
|
||||
import net from 'node:net'
|
||||
import { unzipSync } from 'node:zlib'
|
||||
import nntpConfig from '#config/nntp'
|
||||
import indexerConfig from '#config/indexer'
|
||||
import SeedFetchPool from '#services/SeedFetchPool'
|
||||
import { parseOverviews } from '#lib/nntp/xover_client'
|
||||
|
||||
test.group('SeedFetchPool', () => {
|
||||
test('a chunk that fails on a dead connection is fetched again on a fresh one', async ({ assert }) => {
|
||||
let connections = 0
|
||||
const server = net.createServer((sock) => {
|
||||
const n = ++connections
|
||||
sock.write('200 welcome\r\n')
|
||||
sock.on('data', (chunk) => {
|
||||
for (const line of chunk.toString().split('\r\n').filter(Boolean)) {
|
||||
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.startsWith('XFEATURE')) sock.write('500 no\r\n')
|
||||
else if (line.startsWith('GROUP')) sock.write('211 1 7 7 a.b.test\r\n')
|
||||
// The first connection dies as the request arrives, like one the server dropped while idle.
|
||||
else if (line.startsWith('XOVER') && n === 1) sock.destroy()
|
||||
else if (line.startsWith('XOVER')) sock.write('224 overviews\r\n7\tsubject\tposter\tdate\t<a@b>\t\t10\t1\r\n.\r\n')
|
||||
}
|
||||
})
|
||||
})
|
||||
await new Promise<void>((resolve) => server.listen(0, '127.0.0.1', resolve))
|
||||
const saved = { host: nntpConfig.host, port: nntpConfig.port, secure: nntpConfig.secure, connections: indexerConfig.seed.nntpConnections }
|
||||
Object.assign(nntpConfig, { host: '127.0.0.1', port: (server.address() as net.AddressInfo).port, secure: false })
|
||||
indexerConfig.seed.nntpConnections = 2
|
||||
try {
|
||||
const raw = await SeedFetchPool.fetch('a.b.test', 7, 7)
|
||||
assert.deepEqual(parseOverviews(unzipSync(raw).toString('utf8')).map(([n]) => n), [7])
|
||||
assert.equal(connections, 2)
|
||||
} finally {
|
||||
SeedFetchPool.closeIdle()
|
||||
Object.assign(nntpConfig, { host: saved.host, port: saved.port, secure: saved.secure })
|
||||
indexerConfig.seed.nntpConnections = saved.connections
|
||||
server.close()
|
||||
}
|
||||
})
|
||||
})
|
||||
Reference in new issue
Block a user