Rebuild: chunks the server's compression cuts short are fetched plain

Eweka sends some ranges' compressed XOVER truncated, the same bytes
every time, while the plain reply is whole (a.b.models 3317215-3367214:
1,675,600 bytes that don't inflate, twice; 50,000 lines plain;
tools/false_end.ts). Not a false end of reply. The worker's corrupt
report removed the chunk and the range was fetched again, with the same
bytes: one range was reported 540 times. (Before this morning's change
the inflate check waited for more data until the 10-minute timeout,
like voyager 1's 'chunk never answered' ranges.)

A reported chunk's start goes into ranges.plain_chunks; the prefetcher
fetches those over one connection without XFEATURE COMPRESS GZIP
(NntpPool.usePlain, opened on demand) and deflates them here. A plain
chunk reported again sets the range aside (status failed, error). The
capture (edge:bench --capture) fetches such a chunk plain too.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

app/services/leases.ts

@@ -67,7 +67,14 @@ class Leases {
     const range = await this.leased(rangeId, nodeId)
     const span = Spool.chunksOf({ start: Number(range.start_id), end: Number(range.end_id) }).find((s) => s.from === from)
     if (span) await Spool.removeChunk(rangeId, span)
-    await db.from('ranges').where('id', rangeId).update({ status: 'planned', node_id: null })
+    const plain: number[] = (range.plain_chunks ?? []).map(Number)
+    if (plain.includes(from)) {
+      // Fetched plain (deflated here) and still not inflating: not the server's compression; set aside.
+      await db.from('ranges').where('id', rangeId).update({ status: 'failed', node_id: null, error: `chunk ${from} doesn't inflate, also fetched plain` })
+      return
+    }
+    // The server's compressed reply is cut short (the same bytes every time): fetched plain next time.
+    await db.rawQuery(`UPDATE ranges SET status = 'planned', node_id = NULL, plain_chunks = array_append(plain_chunks, ?::bigint) WHERE id = ?`, [from, rangeId])
   }
 
   /** A stopping worker gives its range back; the next one resumes it. */

app/services/nntp_pool.ts

@@ -75,7 +75,41 @@ class NntpPool {
   }
 
   /** Closes every connection, also those in use: their commands fail. */
+  private plain: Promise<XoverClient> | null = null
+  private plainQueue: Promise<unknown> = Promise.resolve()
+
+  /**
+   * One connection without XFEATURE COMPRESS GZIP, opened when first needed,
+   * one command at a time: for chunks whose compressed reply the server
+   * cuts short (Leases.corrupt). Its replies are deflated here.
+   */
+  async usePlain<T>(work: (client: XoverClient) => Promise<T>): Promise<T> {
+    const run = this.plainQueue.then(async () => {
+      this.plain ??= 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,
+      })
+      const client = await this.plain
+      try {
+        return await work(client)
+      } catch (error) {
+        this.plain = null
+        client.close()
+        throw error
+      }
+    })
+    this.plainQueue = run.catch(() => {})
+    return run
+  }
+
   async close() {
+    void this.plain?.then((c) => c.close()).catch(() => {})
+    this.plain = null
     for (const client of [...this.idle.splice(0), ...this.busy]) client.close()
     this.busy.clear()
     this.open = 0

app/services/prefetcher.ts

@@ -7,6 +7,8 @@ import Spool, { type ChunkSpan } from '#services/spool'
 interface Target {
   id: number
   group: string
+  /** Chunk starts to fetch without compression (ranges.plain_chunks). */
+  plain: Set<number>
   /** Chunks not spooled yet, in order. */
   missing: ChunkSpan[]
   /** All of the range's chunks, from its start. */
@@ -76,10 +78,11 @@ export class Prefetcher {
       const { target, span } = next
       try {
         const fetchedAt = Date.now()
-        const data = await NntpPool.use(async (client) => {
+        const fetch = async (client: Parameters<Parameters<typeof NntpPool.use>[0]>[0]) => {
           await client.group(target.group)
           return client.xoverRaw(span.from, span.to)
-        })
+        }
+        const data = target.plain.has(span.from) ? await NntpPool.usePlain(fetch) : await NntpPool.use(fetch)
         this.stats.chunks++
         this.stats.articles += span.to - span.from + 1
         this.stats.fetchMs += Date.now() - fetchedAt
@@ -134,7 +137,7 @@ export class Prefetcher {
       this.spoolBytes = await Spool.size()
     }
     const mine = (q: any) => (this.shard.count > 1 ? q.whereRaw('r.id % ? = ?', [this.shard.count, this.shard.index]) : q)
-    const rows = (q: any) => mine(q.from('ranges as r').join('groups as g', 'g.id', 'r.group_id').select('r.id', 'g.name as group', 'r.start_id', 'r.end_id', 'r.next_id'))
+    const rows = (q: any) => mine(q.from('ranges as r').join('groups as g', 'g.id', 'r.group_id').select('r.id', 'g.name as group', 'r.start_id', 'r.end_id', 'r.next_id', 'r.plain_chunks'))
     // Leased ranges, checked against the spool's files whatever spooled_at says: a worker waits for them.
     const leased = await rows(db.query()).where('r.status', 'leased').orderBy('r.leased_at')
     // Planned ranges ahead: those started already, then new ones up to SPOOL_AHEAD_PER_LEASE per lease.
@@ -154,7 +157,7 @@ export class Prefetcher {
       for (const span of Spool.chunksOf({ start, end }, Number(row.next_id))) {
         if (!this.claimed.has(`${row.id}:${span.from}`) && !(await Spool.has(Number(row.id), span))) missing.push(span)
       }
-      if (missing.length > 0) targets.push({ id: Number(row.id), group: row.group, missing, total })
+      if (missing.length > 0) targets.push({ id: Number(row.id), group: row.group, plain: new Set((row.plain_chunks ?? []).map(Number)), missing, total })
     }
     return targets
   }

commands/edge_bench.ts

@@ -78,7 +78,17 @@ export default class EdgeBench extends BaseCommand {
     for (let from = this.start; from <= end; from += this.chunk, index++) {
       const to = Math.min(from + this.chunk - 1, end)
       const started = Date.now()
-      const raw = await client.xoverRaw(from, to)
+      let raw = await client.xoverRaw(from, to)
+      // Eweka cuts some ranges' compressed reply short (the same bytes every time): fetched plain then.
+      try {
+        unzipSync(raw)
+      } catch {
+        const plain = await XoverClient.connect({ host: env.get('NNTP_HOST')!, port: env.get('NNTP_PORT')!, user: env.get('NNTP_USER'), password: env.get('NNTP_PASSWORD'), secure: env.get('NNTP_SECURE', true), compress: false, timeoutMs: 300_000 })
+        await plain.group(this.group)
+        raw = await plain.xoverRaw(from, to)
+        plain.close()
+        this.logger.warning(`chunk ${index}: the compressed reply didn't inflate; fetched plain`)
+      }
       await writeFile(join(this.dir, `${String(index).padStart(4, '0')}-${from}-${to}.gz`), raw)
       this.logger.info(`chunk ${index}: ${from}–${to}, ${(raw.length / 1e6).toFixed(1)} MB in ${((Date.now() - started) / 1000).toFixed(1)} s`)
     }

database/migrations/1770000001500_ranges_plain_chunks.ts

@@ -0,0 +1,18 @@
+import { BaseSchema } from '@adonisjs/lucid/schema'
+
+/**
+ * Chunks of a range to fetch without compression: Eweka sends some ranges'
+ * compressed XOVER truncated, the same bytes every time, while the plain
+ * reply is whole (LOG.md 2026-10-09 (11)). A worker reports such a chunk;
+ * its start goes here and the prefetcher fetches it plain.
+ */
+export default class extends BaseSchema {
+  async up() {
+    this.schema.raw(`ALTER TABLE ranges ADD COLUMN IF NOT EXISTS plain_chunks bigint[] NOT NULL DEFAULT '{}'`)
+    this.schema.raw(`ALTER TABLE ranges ADD COLUMN IF NOT EXISTS error text`)
+  }
+
+  async down() {
+    this.schema.raw(`ALTER TABLE ranges DROP COLUMN IF EXISTS plain_chunks, DROP COLUMN IF EXISTS error`)
+  }
+}

tests/functional/node_api.spec.ts

@@ -98,10 +98,19 @@ test.group('Node API', (group) => {
       await Promise.all(Spool.chunksOf({ start: 1, end: 300 }).map((span) => Spool.has(rangeId, span))),
       [true, false, true]
     )
-    const range = await db.from('ranges').where('id', rangeId).first()
+    let range = await db.from('ranges').where('id', rangeId).first()
     assert.equal(range.status, 'planned')
     assert.isNull(range.node_id)
+    // Fetched plain next time (the server's compressed reply was cut short).
+    assert.deepEqual(range.plain_chunks.map(Number), [101])
     // No longer this node's: a second report is refused.
     ;(await client.post(`/api/v1/node/ranges/${rangeId}/corrupt`).bearerToken(token).json({ from: 101 })).assertStatus(409)
+    // Leased again and the plain chunk still doesn't inflate: the range is set aside, not leased in a loop.
+    await Spool.write(rangeId, { from: 101, to: 200 }, chunk(111))
+    ;(await client.post('/api/v1/node/lease').bearerToken(token)).assertStatus(200)
+    ;(await client.post(`/api/v1/node/ranges/${rangeId}/corrupt`).bearerToken(token).json({ from: 101 })).assertStatus(200)
+    range = await db.from('ranges').where('id', rangeId).first()
+    assert.equal(range.status, 'failed')
+    assert.include(range.error, 'also fetched plain')
   })
 })

tools/false_end.ts

@@ -0,0 +1,33 @@
+/**
+ * Whether a compressed XOVER reply that doesn't inflate is the server's: the
+ * range fetched raw twice (compared with a saved copy), and once uncompressed.
+ *
+ *   node --env-file=.env --import=@poppinss/ts-exec tools/false_end.ts <group> <from> <to> [saved.gz]
+ */
+import { readFileSync } from 'node:fs'
+import { unzipSync } from 'node:zlib'
+import { XoverClient } from '#lib/nntp/index'
+
+const [group, from, to, saved] = process.argv.slice(2)
+const env = process.env
+const connect = (compress: boolean) =>
+  XoverClient.connect({ host: env.NNTP_HOST!, port: Number(env.NNTP_PORT ?? 563), user: env.NNTP_USER, password: env.NNTP_PASSWORD, secure: (env.NNTP_SECURE ?? 'true') !== 'false', compress, timeoutMs: 120_000 })
+const gz = await connect(true)
+await gz.group(group)
+for (let i = 0; i < 2; i++) {
+  const raw = await gz.xoverRaw(Number(from), Number(to))
+  let inflates = 'inflates'
+  try {
+    unzipSync(raw)
+  } catch (e: any) {
+    inflates = `doesn't inflate (${e.message})`
+  }
+  const same = saved ? (Buffer.compare(raw, readFileSync(saved)) === 0 ? ', same bytes as saved' : ', differs from saved') : ''
+  console.log(`compressed #${i + 1}: ${raw.length} bytes, ${inflates}${same}`)
+}
+gz.close()
+const plain = await connect(false)
+await plain.group(group)
+const lines = await plain.xover(Number(from), Number(to))
+console.log(`plain: ${lines.length} overview lines`)
+plain.close()