From b7ae8267c60abb35fdd244849779b178c6e29c40 Mon Sep 17 00:00:00 2001 From: Daan Meijer Date: Mon, 5 Oct 2026 21:16:39 +0200 Subject: [PATCH] Seed relay: a stalled compressed XOVER is fetched again uncompressed The four ranges kept failing on fresh connections too: in gzip mode Eweka stops some chunks partway (a.b.ath 49701720000: 19,672 of 20,000 overviews, the last line cut, then silence), while the same chunk without compression comes in 0.3 s. A failed relay chunk is now retried once without compression on a connection of its own (dead pooled connections are covered by the same retry). Co-Authored-By: Claude Opus 5.5 --- app/services/SeedFetchPool.ts | 46 +++++++++++++++++++++--------- tests/unit/seed_fetch_pool.spec.ts | 7 +++-- 2 files changed, 38 insertions(+), 15 deletions(-) diff --git a/app/services/SeedFetchPool.ts b/app/services/SeedFetchPool.ts index 4026d9d..9fbee76 100644 --- a/app/services/SeedFetchPool.ts +++ b/app/services/SeedFetchPool.ts @@ -30,15 +30,45 @@ class SeedFetchPool { return indexerConfig.seed.nntpConnections } - /** Retried once on a fresh connection: a failure is nearly always a dead connection, not the articles. */ + /** + * A failure is retried once without compression, on a connection of its + * own: either the pooled connection was dead (dropped while idle), or the + * server stalled mid-reply in compressed mode. Eweka does that for some + * chunks: it stops after most of the stream and never sends the rest + * (a.b.ath 49701720000: 19,672 of 20,000 overviews, then silence), while + * the same chunk uncompressed comes in 0.3 s (LOG.md 2026-10-05 (11)). + */ async fetch(group: string, from: number, to: number): Promise { try { return await this.fetchOnce(group, from, to) } catch { - return this.fetchOnce(group, from, to) + return this.fetchPlain(group, from, to) } } + private async fetchPlain(group: string, from: number, to: number): Promise { + const client = await this.connect(false) + try { + await client.group(group) + return await client.xoverRaw(from, to) + } finally { + client.close() + } + } + + private connect(compress: boolean) { + return XoverClient.connect({ + host: nntpConfig.host, + port: nntpConfig.port, + secure: nntpConfig.secure, + servername: nntpConfig.tlsServername, + user: nntpConfig.user, + password: nntpConfig.password?.release(), + timeoutMs: indexerConfig.seed.nntpTimeoutMs, + compress, + }) + } + private async fetchOnce(group: string, from: number, to: number): Promise { const slot = await this.acquire() let ok = false @@ -75,17 +105,7 @@ class SeedFetchPool { if (this.open < this.size) { this.open++ try { - const client = await XoverClient.connect({ - host: nntpConfig.host, - port: nntpConfig.port, - secure: nntpConfig.secure, - servername: nntpConfig.tlsServername, - user: nntpConfig.user, - password: nntpConfig.password?.release(), - timeoutMs: indexerConfig.seed.nntpTimeoutMs, - compress: true, - }) - return { client, usedAt: Date.now() } + return { client: await this.connect(true), usedAt: Date.now() } } catch (error) { this.open-- this.waiters.shift()?.() diff --git a/tests/unit/seed_fetch_pool.spec.ts b/tests/unit/seed_fetch_pool.spec.ts index 8fe9417..bbe87ab 100644 --- a/tests/unit/seed_fetch_pool.spec.ts +++ b/tests/unit/seed_fetch_pool.spec.ts @@ -7,8 +7,9 @@ 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 }) => { + test('a chunk that fails (dead connection, stalled compressed reply) is fetched again uncompressed', async ({ assert }) => { let connections = 0 + const compressAsked: Record = {} const server = net.createServer((sock) => { const n = ++connections sock.write('200 welcome\r\n') @@ -16,7 +17,7 @@ test.group('SeedFetchPool', () => { 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('XFEATURE')) (compressAsked[n] = true), 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() @@ -32,6 +33,8 @@ test.group('SeedFetchPool', () => { 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) + // The pooled connection asked for gzip; the retry didn't. + assert.deepEqual(compressAsked, { 1: true }) } finally { SeedFetchPool.closeIdle() Object.assign(nntpConfig, { host: saved.host, port: saved.port, secure: saved.secure })