diff --git a/commands/nntp_probe.ts b/commands/nntp_probe.ts index def2aec..58bc852 100644 --- a/commands/nntp_probe.ts +++ b/commands/nntp_probe.ts @@ -95,7 +95,7 @@ export default class NntpProbe extends BaseCommand { } catch (error: any) { const timedOut = /timed out/i.test(error.message) this.logger.warning( - ` XOVER ${from}-${to}: ${error.message} after ${Date.now() - startedAt} ms${client ? `, ${client.buffered} bytes buffered` : ''}` + ` XOVER ${from}-${to}: ${error.message} after ${Date.now() - startedAt} ms${client ? `, ${client.buffered} bytes buffered: ${JSON.stringify(client.bufferedHead().toString('latin1'))} (${client.bufferedHead().toString('hex')})` : ''}` ) return timedOut } finally { diff --git a/lib/nntp/xover_client.ts b/lib/nntp/xover_client.ts index 529ec00..520f03d 100644 --- a/lib/nntp/xover_client.ts +++ b/lib/nntp/xover_client.ts @@ -54,6 +54,10 @@ export class XoverClient { get buffered() { return this.length } + /** The first `n` bytes received and not consumed yet (what a reply that stalled left; nntp:probe). */ + bufferedHead(n = 32): Buffer { + return this.buffer().subarray(0, n) + } private lastDataAt = 0 private longestGap = 0 @@ -85,7 +89,11 @@ export class XoverClient { static async connect(options: XoverClientOptions): Promise { const sock = options.secure - ? tls.connect({ host: options.host, port: options.port, servername: options.servername || options.host }) + ? tls.connect({ + host: options.host, + port: options.port, + servername: options.servername || options.host, + }) : net.connect({ host: options.host, port: options.port }) const client = new XoverClient(sock, options.timeoutMs ?? 120_000) // The greeting and login are short: a server holding a connection back (too many open) fails it in 30 s, not timeoutMs. @@ -94,10 +102,14 @@ export class XoverClient { await client.expect(await client.line(), ['200', '201']) if (options.user) { const reply = await client.command(`AUTHINFO USER ${options.user}`) - if (reply.startsWith('381')) await client.expect(await client.command(`AUTHINFO PASS ${options.password ?? ''}`), ['281']) + if (reply.startsWith('381')) + await client.expect(await client.command(`AUTHINFO PASS ${options.password ?? ''}`), [ + '281', + ]) else await client.expect(reply, ['281']) } - if (options.compress) client.compressed = (await client.command('XFEATURE COMPRESS GZIP')).startsWith('290') + if (options.compress) + client.compressed = (await client.command('XFEATURE COMPRESS GZIP')).startsWith('290') } catch (error) { client.close() throw error @@ -127,7 +139,9 @@ export class XoverClient { const reply = await this.command(`LIST ACTIVE ${pattern}`) await this.expect(reply, ['215']) // With XFEATURE COMPRESS GZIP on, Eweka compresses this list too ("215 NewsGroups Follow [COMPRESS=GZIP]"). - const body = reply.includes('COMPRESS=GZIP') ? (await this.compressedRaw(true)).text : await this.plainBody() + const body = reply.includes('COMPRESS=GZIP') + ? (await this.compressedRaw(true)).text + : await this.plainBody() return body .split('\r\n') .filter(Boolean) @@ -154,7 +168,11 @@ export class XoverClient { return deflateAsync(Buffer.from(await this.plainBody()), { level: 1 }) } finally { this.measuring = false - this.lastTiming = { replyMs: repliedAt - sentAt, bodyMs: performance.now() - repliedAt, longestGapMs: this.longestGap } + this.lastTiming = { + replyMs: repliedAt - sentAt, + bodyMs: performance.now() - repliedAt, + longestGapMs: this.longestGap, + } } }) } @@ -180,7 +198,9 @@ export class XoverClient { const reply = await this.command(`XOVER ${from}-${to}`) if (reply.startsWith('420') || reply.startsWith('423')) return [] await this.expect(reply, ['224']) - const data = reply.includes('COMPRESS=GZIP') ? await this.compressedBody() : await this.plainBody() + const data = reply.includes('COMPRESS=GZIP') + ? await this.compressedBody() + : await this.plainBody() return parseOverviews(data) } @@ -199,7 +219,8 @@ export class XoverClient { } private async expect(reply: string, codes: string[]) { - if (!codes.some((code) => reply.startsWith(code))) throw new Error(`NNTP: unexpected reply "${reply}"`) + if (!codes.some((code) => reply.startsWith(code))) + throw new Error(`NNTP: unexpected reply "${reply}"`) } private buffer(): Buffer { @@ -226,7 +247,10 @@ export class XoverClient { private async line(): Promise { while (true) { const at = this.buffer().indexOf('\r\n') - if (at !== -1) return this.take(at + 2).subarray(0, at).toString('latin1') + if (at !== -1) + return this.take(at + 2) + .subarray(0, at) + .toString('latin1') await this.more() } } @@ -267,7 +291,9 @@ export class XoverClient { */ async bodies(messageIds: string[]): Promise<(Buffer | null)[]> { return this.timed(async () => { - this.sock.write(messageIds.map((m) => `BODY ${m.startsWith('<') ? m : `<${m}>`}\r\n`).join('')) + this.sock.write( + messageIds.map((m) => `BODY ${m.startsWith('<') ? m : `<${m}>`}\r\n`).join('') + ) const out: (Buffer | null)[] = [] for (let i = 0; i < messageIds.length; i++) { const reply = await this.line() @@ -309,7 +335,11 @@ export class XoverClient { return '' } const at = all.indexOf(TERMINATOR, Math.max(0, searched - TERMINATOR.length)) - if (at !== -1) return this.take(at + TERMINATOR.length).subarray(0, at + 2).toString('utf8').replace(/^\.\./gm, '.') + if (at !== -1) + return this.take(at + TERMINATOR.length) + .subarray(0, at + 2) + .toString('utf8') + .replace(/^\.\./gm, '.') searched = all.length await this.more() }