nntp:probe: shows the bytes a stalled reply left

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>

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 {

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<XoverClient> {
     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<string> {
     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()
     }