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 <noreply@anthropic.com>
This commit is contained in:
1 parent
028dd90a6f
commit
b7ae8267c6
2 files changed
+38
-15
No files matched your search
@@ -30,15 +30,45 @@ class SeedFetchPool {
|
|||||||
return indexerConfig.seed.nntpConnections
|
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<Buffer> {
|
async fetch(group: string, from: number, to: number): Promise<Buffer> {
|
||||||
try {
|
try {
|
||||||
return await this.fetchOnce(group, from, to)
|
return await this.fetchOnce(group, from, to)
|
||||||
} catch {
|
} catch {
|
||||||
return this.fetchOnce(group, from, to)
|
return this.fetchPlain(group, from, to)
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
private async fetchPlain(group: string, from: number, to: number): Promise<Buffer> {
|
||||||
|
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<Buffer> {
|
private async fetchOnce(group: string, from: number, to: number): Promise<Buffer> {
|
||||||
const slot = await this.acquire()
|
const slot = await this.acquire()
|
||||||
let ok = false
|
let ok = false
|
||||||
@@ -75,17 +105,7 @@ class SeedFetchPool {
|
|||||||
if (this.open < this.size) {
|
if (this.open < this.size) {
|
||||||
this.open++
|
this.open++
|
||||||
try {
|
try {
|
||||||
const client = await XoverClient.connect({
|
return { client: await this.connect(true), usedAt: Date.now() }
|
||||||
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() }
|
|
||||||
} catch (error) {
|
} catch (error) {
|
||||||
this.open--
|
this.open--
|
||||||
this.waiters.shift()?.()
|
this.waiters.shift()?.()
|
||||||
|
|||||||
@@ -7,8 +7,9 @@ import SeedFetchPool from '#services/SeedFetchPool'
|
|||||||
import { parseOverviews } from '#lib/nntp/xover_client'
|
import { parseOverviews } from '#lib/nntp/xover_client'
|
||||||
|
|
||||||
test.group('SeedFetchPool', () => {
|
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
|
let connections = 0
|
||||||
|
const compressAsked: Record<number, boolean> = {}
|
||||||
const server = net.createServer((sock) => {
|
const server = net.createServer((sock) => {
|
||||||
const n = ++connections
|
const n = ++connections
|
||||||
sock.write('200 welcome\r\n')
|
sock.write('200 welcome\r\n')
|
||||||
@@ -16,7 +17,7 @@ test.group('SeedFetchPool', () => {
|
|||||||
for (const line of chunk.toString().split('\r\n').filter(Boolean)) {
|
for (const line of chunk.toString().split('\r\n').filter(Boolean)) {
|
||||||
if (line.startsWith('AUTHINFO USER')) sock.write('381 more\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.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')
|
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.
|
// 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') && n === 1) sock.destroy()
|
||||||
@@ -32,6 +33,8 @@ test.group('SeedFetchPool', () => {
|
|||||||
const raw = await SeedFetchPool.fetch('a.b.test', 7, 7)
|
const raw = await SeedFetchPool.fetch('a.b.test', 7, 7)
|
||||||
assert.deepEqual(parseOverviews(unzipSync(raw).toString('utf8')).map(([n]) => n), [7])
|
assert.deepEqual(parseOverviews(unzipSync(raw).toString('utf8')).map(([n]) => n), [7])
|
||||||
assert.equal(connections, 2)
|
assert.equal(connections, 2)
|
||||||
|
// The pooled connection asked for gzip; the retry didn't.
|
||||||
|
assert.deepEqual(compressAsked, { 1: true })
|
||||||
} finally {
|
} finally {
|
||||||
SeedFetchPool.closeIdle()
|
SeedFetchPool.closeIdle()
|
||||||
Object.assign(nntpConfig, { host: saved.host, port: saved.port, secure: saved.secure })
|
Object.assign(nntpConfig, { host: saved.host, port: saved.port, secure: saved.secure })
|
||||||
|
|||||||
Reference in new issue
Block a user