TruncatedReply: a compressed reply the server cut at a ".\r\n" fails in 2 s; ranges:dates samples plain

Eweka stops a compressed XOVER reply at the first ".\r\n" in its own
output (78 01 2e 0d 0a and nothing more), which some small replies begin
with; the text path waited for more until the reply timeout. ranges:dates
uses plain connections of its own instead of the pool.

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

REBUILD.md

@@ -247,7 +247,10 @@ corpus filtered this way no longer serves a sweep of those two rules.
 Windows by posting time (2026-10-25; voyager 3's test data, LOG.md (104)):
 `ranges:dates` samples when each active group's articles were posted (one
 XOVER of 32 articles every 500M articles of the group, 16 … 1000 samples,
-the median Date of each slice, kept monotone; `group_dates`) and estimates
+the median Date of each slice, kept monotone; `group_dates`; over plain
+connections: Eweka cuts a compressed reply at the first ".\r\n" in its
+output, which some small replies begin with, `TruncatedReply`, LOG.md
+2026-10-25 (105)) and estimates
 every range's posting date from the curve (`ranges.est_date`; the edges of
 a window are off by up to one sample interval, three weeks in boneless).
 `ranges:capture --from 2025-01-01 --to 2026-01-01` marks every active

commands/ranges_dates.ts

@@ -1,8 +1,9 @@
 import { BaseCommand, flags } from '@adonisjs/core/ace'
 import type { CommandOptions } from '@adonisjs/core/types/ace'
 import db from '@adonisjs/lucid/services/db'
+import env from '#start/env'
 import coordinatorConfig from '#config/coordinator'
-import NntpPool from '#services/nntp_pool'
+import { XoverClient } from '#lib/nntp/index'
 import { NntpShare } from '#services/nntp_shares'
 import { estimateDate, sampleGroup, type DateSample } from '#services/group_dates'
 
@@ -39,13 +40,9 @@ export default class RangesDates extends BaseCommand {
   declare estimateOnly: boolean
 
   async run() {
-    try {
-      if (!this.estimateOnly) await this.sample()
-      await this.estimate()
-      await this.report()
-    } finally {
-      await NntpPool.close()
-    }
+    if (!this.estimateOnly) await this.sample()
+    await this.estimate()
+    await this.report()
   }
 
   private groups(missing = this.missing) {
@@ -62,20 +59,26 @@ export default class RangesDates extends BaseCommand {
     const every = this.every ?? 500_000_000
     const connections = this.connections ?? 6
     const groups = await this.groups()
+    // Connections of its own, without compression: Eweka cuts a compressed reply at the first ".\r\n" in its
+    // output, which 16 of 14k small XOVERs hit (TruncatedReply; LOG.md 2026-10-25 (105)); plain, every one
+    // answers, and 32 lines are a few kB. Taken from the account as a share (the coordinator and the prober make
+    // room for them), as many as asked for.
+    const workers = Math.max(1, Math.min(connections, groups.length))
     const share = new NntpShare(
       'dates',
       coordinatorConfig.nntpShares.coordinatorPriority + 1,
-      1,
-      connections
-    )
-    const first = await NntpPool.share(share)
-    const sharing = setInterval(
-      () =>
-        NntpPool.share(share).catch((error) => this.logger.warning(`NNTP share: ${error.message}`)),
-      coordinatorConfig.nntpShares.everyMs
+      workers,
+      workers
     )
+    let open = 0
+    const sync = () =>
+      share
+        .sync(workers, open)
+        .catch((error) => this.logger.warning(`NNTP share: ${error.message}`))
+    await sync()
+    const sharing = setInterval(sync, coordinatorConfig.nntpShares.everyMs)
     this.logger.info(
-      `Sampling ${groups.length} groups over ${first.granted} connections, a sample every ${every / 1e6}M articles.`
+      `Sampling ${groups.length} groups over ${workers} plain connections, a sample every ${every / 1e6}M articles.`
     )
     const startedAt = Date.now()
     let done = 0
@@ -83,12 +86,32 @@ export default class RangesDates extends BaseCommand {
     let samples = 0
     const queue = [...groups]
     const worker = async () => {
+      let client: XoverClient | null = null
+      const use = async <T>(work: (c: XoverClient) => Promise<T>): Promise<T> => {
+        if (!client) {
+          client = await XoverClient.connect({
+            host: env.get('NNTP_HOST')!,
+            port: env.get('NNTP_PORT', 563),
+            user: env.get('NNTP_USER'),
+            password: env.get('NNTP_PASSWORD'),
+            secure: env.get('NNTP_SECURE', true),
+            compress: false,
+            timeoutMs: coordinatorConfig.nntpTimeoutMs,
+          })
+          open++
+        }
+        try {
+          return await work(client)
+        } catch (error) {
+          client.close()
+          client = null
+          open--
+          throw error
+        }
+      }
       for (let g = queue.shift(); g; g = queue.shift()) {
         try {
-          const result = await sampleGroup((work) => NntpPool.use(work), g.name, {
-            every,
-            size: 32,
-          })
+          const result = await sampleGroup(use, g.name, { every, size: 32 })
           for (const f of result.failed) this.logger.warning(`${g.name} at ${f}`)
           await db.transaction(async (trx) => {
             await trx.from('group_dates').where('group_id', g.id).delete()
@@ -116,13 +139,16 @@ export default class RangesDates extends BaseCommand {
             `${done} of ${groups.length} groups: ${samples} samples, ${commands} commands, ${((Date.now() - startedAt) / 1000).toFixed(0)} s.`
           )
       }
+      if (client) {
+        ;(client as XoverClient).close()
+        open--
+      }
     }
     try {
-      await Promise.all(
-        Array.from({ length: Math.max(1, Math.min(connections, groups.length)) }, worker)
-      )
+      await Promise.all(Array.from({ length: workers }, worker))
     } finally {
       clearInterval(sharing)
+      await share.sync(0, 0).catch(() => {})
     }
     this.logger.info(
       `Sampled ${groups.length} groups: ${samples} samples, ${commands} commands, ${((Date.now() - startedAt) / 1000).toFixed(0)} s.`

lib/nntp/index.ts

@@ -1,4 +1,11 @@
-export { XoverClient, MissingArticle, parseOverviews, type Overview, type XoverClientOptions } from './xover_client.js'
+export {
+  XoverClient,
+  MissingArticle,
+  TruncatedReply,
+  parseOverviews,
+  type Overview,
+  type XoverClientOptions,
+} from './xover_client.js'
 export { parseXrefGroups } from './xref.js'
 export { parseGroupResponse, type GroupInfo } from './group.js'
 export { hedge, type Request, type Hedged } from './hedge.js'

lib/nntp/xover_client.ts

@@ -36,6 +36,14 @@ const DOT_LINE = Buffer.from('.\r\n')
  */
 /** The server doesn't have the article (430): taken down or expired. */
 export class MissingArticle extends Error {}
+/**
+ * A compressed reply the server cut short: it ends its compressed output at the first ".\r\n" in it, as if that
+ * were the terminator, and sends nothing more (Eweka; "78 01 2e 0d 0a" for 16 of 14k small XOVERs, LOG.md
+ * 2026-10-25 (105)). The same request without compression answers.
+ */
+export class TruncatedReply extends Error {}
+/** How long a compressed reply that doesn't inflate may stay silent before it counts as cut short. */
+const TRUNCATED_MS = 2_000
 
 export class XoverClient {
   private chunks: Buffer[] = []
@@ -244,6 +252,21 @@ export class XoverClient {
     if (this.failure && this.length === 0) throw this.failure
   }
 
+  /** As more(), but false when nothing came within `ms`. */
+  private async moreWithin(ms: number): Promise<boolean> {
+    if (this.failure) throw this.failure
+    const came = await new Promise<boolean>((resolve) => {
+      const timer = setTimeout(() => resolve(false), ms)
+      this.wake = () => {
+        clearTimeout(timer)
+        resolve(true)
+      }
+    })
+    this.wake = undefined
+    if (this.failure && this.length === 0) throw this.failure
+    return came
+  }
+
   private async line(): Promise<string> {
     while (true) {
       const at = this.buffer().indexOf('\r\n')
@@ -389,7 +412,14 @@ export class XoverClient {
           this.take(all.length)
           return { raw, text: inflated.toString('utf8') }
         } catch {
-          // ".\r\n" inside the compressed data: wait for more.
+          // ".\r\n" inside the compressed data: wait for more, but not for the reply timeout: the server may have
+          // cut its output there (TruncatedReply), and then nothing comes. Raw, the chunk is returned as it is
+          // and reported by the worker; the text is wanted here.
+          if (!(await this.moreWithin(TRUNCATED_MS))) {
+            this.take(all.length)
+            throw new TruncatedReply(`NNTP: compressed reply cut short after ${all.length} bytes`)
+          }
+          continue
         }
       }
       await this.more()

tests/unit/nntp/xover_client.spec.ts

@@ -1,7 +1,7 @@
 import { test } from '@japa/runner'
 import net from 'node:net'
 import { deflateSync } from 'node:zlib'
-import { MissingArticle, XoverClient, parseOverviews } from '#lib/nntp/index'
+import { MissingArticle, TruncatedReply, XoverClient, parseOverviews } from '#lib/nntp/index'
 import { unzipSync } from 'node:zlib'
 
 const LINES = [
@@ -27,10 +27,15 @@ function server(): Promise<{ port: number; close: () => void }> {
           pending = pending.slice(at + 2)
           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 === 'XFEATURE COMPRESS GZIP') (gzip = true), sock.write('290 Feature Enabled\r\n')
+          else if (line === 'XFEATURE COMPRESS GZIP')
+            ((gzip = true), sock.write('290 Feature Enabled\r\n'))
           else if (line.startsWith('GROUP')) sock.write('211 2 101 102 a.b.test\r\n')
           else if (line === 'XOVER 1-1') sock.write('420 No Articles Selected\r\n')
-          else if (line.startsWith('XOVER')) {
+          else if (line === 'XOVER 5-5' && gzip) {
+            // As Eweka cuts a compressed reply whose output has ".\r\n" early: the zlib header, then nothing.
+            sock.write('224 Overview Information Follows [COMPRESS=GZIP]\r\n')
+            sock.write(Buffer.from([0x78, 0x01, 0x2e, 0x0d, 0x0a]))
+          } else if (line.startsWith('XOVER')) {
             const body = Buffer.from(LINES.join('\r\n') + '\r\n')
             if (gzip) {
               sock.write('224 Overview Information Follows [COMPRESS=GZIP]\r\n')
@@ -53,19 +58,32 @@ function server(): Promise<{ port: number; close: () => void }> {
           else if (line.startsWith('BODY')) {
             sock.write('222 0 <one@test> body\r\n')
             // Raw bytes, a dot-stuffed line among them.
-            send(Buffer.concat([Buffer.from([0x3d, 0x79, 0xff, 0x00, 0x0d, 0x0a]), Buffer.from('..stuffed\r\nlast\r\n.\r\n')]))
+            send(
+              Buffer.concat([
+                Buffer.from([0x3d, 0x79, 0xff, 0x00, 0x0d, 0x0a]),
+                Buffer.from('..stuffed\r\nlast\r\n.\r\n'),
+              ])
+            )
           } else if (line === 'QUIT') sock.end('205 bye\r\n')
         }
       })
     })
-    srv.listen(0, '127.0.0.1', () => resolve({ port: (srv.address() as net.AddressInfo).port, close: () => srv.close() }))
+    srv.listen(0, '127.0.0.1', () =>
+      resolve({ port: (srv.address() as net.AddressInfo).port, close: () => srv.close() })
+    )
   })
 }
 
 test.group('nntp / XoverClient', () => {
   test('reads a body by message-id as bytes, raw; 430 is MissingArticle', async ({ assert }) => {
     const { port, close } = await server()
-    const client = await XoverClient.connect({ host: '127.0.0.1', port, user: 'u', password: 'p', compress: false })
+    const client = await XoverClient.connect({
+      host: '127.0.0.1',
+      port,
+      user: 'u',
+      password: 'p',
+      compress: false,
+    })
     try {
       const body = await client.body('one@test')
       assert.deepEqual([...body.subarray(0, 6)], [0x3d, 0x79, 0xff, 0x00, 0x0d, 0x0a])
@@ -81,7 +99,13 @@ test.group('nntp / XoverClient', () => {
 
   test('reads several bodies pipelined, in order, a missing one as null', async ({ assert }) => {
     const { port, close } = await server()
-    const client = await XoverClient.connect({ host: '127.0.0.1', port, user: 'u', password: 'p', compress: false })
+    const client = await XoverClient.connect({
+      host: '127.0.0.1',
+      port,
+      user: 'u',
+      password: 'p',
+      compress: false,
+    })
     try {
       const bodies = await client.bodies(['one@test', '<gone@test>', '<one@test>'])
       assert.equal(bodies.length, 3)
@@ -105,14 +129,25 @@ test.group('nntp / XoverClient', () => {
   })
 
   for (const compress of [false, true]) {
-    test(`fetches overviews ${compress ? 'gzip-compressed' : 'plain'}, and an empty range`, async ({ assert }) => {
+    test(`fetches overviews ${compress ? 'gzip-compressed' : 'plain'}, and an empty range`, async ({
+      assert,
+    }) => {
       const { port, close } = await server()
-      const client = await XoverClient.connect({ host: '127.0.0.1', port, user: 'u', password: 'p', compress })
+      const client = await XoverClient.connect({
+        host: '127.0.0.1',
+        port,
+        user: 'u',
+        password: 'p',
+        compress,
+      })
       try {
         assert.equal(client.compressed, compress)
         assert.deepEqual(await client.group('a.b.test'), { count: 2, low: 101, high: 102 })
         const overviews = await client.xover(101, 102)
-        assert.deepEqual(overviews.map(([n]) => n), [101, 102])
+        assert.deepEqual(
+          overviews.map(([n]) => n),
+          [101, 102]
+        )
         assert.equal(overviews[1][1].subject, 'Show.S01E01 - "a.mkv" yEnc (2/2)')
         assert.deepEqual(await client.xover(1, 1), [])
         assert.lengthOf(await client.xover(101, 102), 2)
@@ -127,9 +162,43 @@ test.group('nntp / XoverClient', () => {
     })
   }
 
-  test('idle time between commands does not count against the reply timeout', async ({ assert }) => {
+  test('a compressed reply cut short at a ".\\r\\n" in its output fails as TruncatedReply, well before the reply timeout', async ({
+    assert,
+  }) => {
     const { port, close } = await server()
-    const client = await XoverClient.connect({ host: '127.0.0.1', port, user: 'u', password: 'p', compress: true, timeoutMs: 200 })
+    const client = await XoverClient.connect({
+      host: '127.0.0.1',
+      port,
+      user: 'u',
+      password: 'p',
+      compress: true,
+      timeoutMs: 60_000,
+    })
+    try {
+      await client.group('a.b.test')
+      const startedAt = Date.now()
+      await assert.rejects(() => client.xover(5, 5), TruncatedReply)
+      assert.isBelow(Date.now() - startedAt, 10_000)
+      // The connection is still usable: the server thinks it answered.
+      assert.lengthOf(await client.xover(101, 102), 2)
+    } finally {
+      client.close()
+      close()
+    }
+  }).timeout(10_000)
+
+  test('idle time between commands does not count against the reply timeout', async ({
+    assert,
+  }) => {
+    const { port, close } = await server()
+    const client = await XoverClient.connect({
+      host: '127.0.0.1',
+      port,
+      user: 'u',
+      password: 'p',
+      compress: true,
+      timeoutMs: 200,
+    })
     try {
       await client.group('a.b.test')
       await new Promise((resolve) => setTimeout(resolve, 500))
@@ -141,13 +210,24 @@ test.group('nntp / XoverClient', () => {
   })
 
   for (const compress of [false, true]) {
-    test(`hands out overviews as one zlib stream (${compress ? 'as the server sent it' : 'deflated here'})`, async ({ assert }) => {
+    test(`hands out overviews as one zlib stream (${compress ? 'as the server sent it' : 'deflated here'})`, async ({
+      assert,
+    }) => {
       const { port, close } = await server()
-      const client = await XoverClient.connect({ host: '127.0.0.1', port, user: 'u', password: 'p', compress })
+      const client = await XoverClient.connect({
+        host: '127.0.0.1',
+        port,
+        user: 'u',
+        password: 'p',
+        compress,
+      })
       try {
         await client.group('a.b.test')
         const raw = await client.xoverRaw(101, 102)
-        assert.deepEqual(parseOverviews(unzipSync(raw).toString('utf8')).map(([n]) => n), [101, 102])
+        assert.deepEqual(
+          parseOverviews(unzipSync(raw).toString('utf8')).map(([n]) => n),
+          [101, 102]
+        )
         assert.lengthOf(parseOverviews(unzipSync(await client.xoverRaw(1, 1)).toString('utf8')), 0)
       } finally {
         client.close()