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
   }