Rebuild: first end-to-end runs; prefetch per chunk, executable rule, wire fixes

End to end on the workstation (coordinator:run, the API, edge:node, real
NNTP): a.b.e-book.french and a.b.cd.image.playstation2.dvdiso ranges.

- Prefetcher: every connection takes the next missing chunk (leased
  ranges first, in order); one connection per range (~26k lines/s) kept
  the worker waiting
- executable rule: releases of bare executables (malware floods named
  after films and software) are dropped; they were 92% of the catalog rows
  of the test ranges, 21% of the segments
- fragments per file (details once) and gzipped batch bodies; the JSON
  limit (1 MB by default) raised to 256 MB, empty strings kept (a rule
  count's key is '')
- release_chunks.batch_from is a bigint (boneless article numbers)
- tests get their own database; edge:bench --store times BatchStore:
  1.3 s for a boneless range (1.24 ms per release chunk)

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

commands/edge_bench.ts

@@ -7,6 +7,8 @@ import env from '#start/env'
 import { XoverClient, parseOverviews } from '#lib/nntp/index'
 import { RangeIndexer } from '#lib/edge/index'
 import { DEFAULT_RULES, type Rules } from '#lib/rules/index'
+import db from '@adonisjs/lucid/services/db'
+import BatchStore from '#services/batch_store'
 
 /**
  * Measures the edge worker on a real range (REBUILD.md: TypeScript or Rust):
@@ -43,6 +45,9 @@ export default class EdgeBench extends BaseCommand {
   @flags.number({ description: 'Chunks without a new part before an open file is given up', default: 5 })
   declare idle: number
 
+  @flags.boolean({ description: 'Also store the batches (BatchStore, the configured database), timed' })
+  declare store: boolean
+
   @flags.string({ description: 'Indexed groups in priority order, comma-separated (owner rule)' })
   declare priorities?: string
 
@@ -92,6 +97,16 @@ export default class EdgeBench extends BaseCommand {
     const counts = new Map<string, number>()
     const keys = new Set<string>()
     let batches = 0
+    let storeMs = 0
+    let rangeId = 0
+    let nodeId = 0
+    if (this.store) {
+      const [g] = await db.table('groups').insert({ name: `bench:${group}:${Date.now()}`, priority: 1_000_000 + (Date.now() % 1_000_000) }).returning('id')
+      const [n] = await db.table('nodes').insert({ name: `bench-${Date.now()}` }).returning('id')
+      nodeId = Number(n.id)
+      const [r] = await db.table('ranges').insert({ group_id: g.id, start_id: start, end_id: end, next_id: start, status: 'leased', node_id: nodeId }).returning('id')
+      rangeId = Number(r.id)
+    }
     const cpuStart = process.cpuUsage()
     const wall = Date.now()
     for (const file of files) {
@@ -114,6 +129,11 @@ export default class EdgeBench extends BaseCommand {
       at = performance.now()
       const batch = indexer.take()
       t.index += performance.now() - at
+      if (this.store) {
+        const storedAt = performance.now()
+        await BatchStore.store(rangeId, nodeId, batch)
+        storeMs += performance.now() - storedAt
+      }
       out.headers += batch.headers
       out.releases += batch.releases.length
       batches++
@@ -138,5 +158,6 @@ export default class EdgeBench extends BaseCommand {
     this.logger.info(`  ${batches} batches of ${this.batchChunks} chunks: ${out.releases} release chunks for ${keys.size} releases (${(out.releases / Math.max(1, keys.size)).toFixed(2)} chunks per release)`)
     this.logger.info(`  kept: ${out.releases} release chunks, ${out.files} files, ${out.segments} segments, payload ${(out.payload / 1e6).toFixed(1)} MB (${(out.payload / Math.max(1, out.segments)).toFixed(1)} B/segment); ${out.candidates} candidates, ${out.fragments} fragments, ${out.samples} samples`)
     for (const [k, v] of [...counts].sort((a, b) => b[1] - a[1]).slice(0, 8)) this.logger.info(`  dropped ${k}: ${v.toLocaleString('en')} articles`)
+    if (this.store) this.logger.info(`  stored ${batches} batches in ${(storeMs / 1000).toFixed(1)} s (${(storeMs / Math.max(1, out.releases)).toFixed(2)} ms per release chunk)`)
   }
 }

commands/edge_node.ts

@@ -1,7 +1,7 @@
 import { BaseCommand, flags } from '@adonisjs/core/ace'
 import type { CommandOptions } from '@adonisjs/core/types/ace'
 import { setTimeout as sleep } from 'node:timers/promises'
-import { unzipSync } from 'node:zlib'
+import { gzipSync, unzipSync } from 'node:zlib'
 import { parseOverviews } from '#lib/nntp/index'
 import { RangeIndexer, decodeFrames, toWire, type EdgeBatch, type Lease } from '#lib/edge/index'
 
@@ -68,10 +68,13 @@ export default class EdgeNode extends BaseCommand {
     const { range, rules } = lease
     const started = Date.now()
     const indexer = new RangeIndexer({ group: range.group, start: range.start, end: range.end, rules })
-    const totals = { headers: 0, releases: 0, candidates: 0 }
+    const totals = { headers: 0, releases: 0, candidates: 0, bytes: 0, postMs: 0 }
     let chunks = 0
     const post = async (batch: EdgeBatch) => {
+      const at = Date.now()
       const stored = await api.batch(range.id, batch)
+      totals.postMs += Date.now() - at
+      totals.bytes += stored.bytes
       totals.headers += batch.headers
       totals.releases += stored.releases
       totals.candidates += stored.candidates
@@ -94,7 +97,8 @@ export default class EdgeNode extends BaseCommand {
       const seconds = (Date.now() - started) / 1000
       this.logger.info(
         `${range.group} ${range.start}–${range.end}: ${totals.headers.toLocaleString('en')} headers in ${seconds.toFixed(0)} s ` +
-          `(${Math.round(totals.headers / seconds).toLocaleString('en')}/s), ${totals.releases} release chunks, ${totals.candidates} probe candidates.`
+          `(${Math.round(totals.headers / seconds).toLocaleString('en')}/s), ${totals.releases} release chunks, ${totals.candidates} probe candidates; ` +
+          `posted ${(totals.bytes / 1e6).toFixed(1)} MB in ${(totals.postMs / 1000).toFixed(1)} s.`
       )
     } catch (error: any) {
       if (error instanceof LeaseLost) {
@@ -144,10 +148,12 @@ class Api {
     return response.body as unknown as AsyncIterable<Uint8Array>
   }
 
-  async batch(rangeId: number, batch: EdgeBatch): Promise<{ releases: number; candidates: number }> {
-    const response = await this.request(`ranges/${rangeId}/batches`, { method: 'POST', body: JSON.stringify(toWire(batch)) })
+  async batch(rangeId: number, batch: EdgeBatch): Promise<{ releases: number; candidates: number; bytes: number }> {
+    // Gzipped: chunk payloads are compressed already, the rest (names, fragments, candidates) compresses well.
+    const body = gzipSync(JSON.stringify(toWire(batch)), { level: 1 })
+    const response = await this.request(`ranges/${rangeId}/batches`, { method: 'POST', body, headers: { 'Content-Encoding': 'gzip' } })
     if (!response.ok) throw new Error(`batch: HTTP ${response.status}`)
-    return response.json() as Promise<{ releases: number; candidates: number }>
+    return { ...((await response.json()) as { releases: number; candidates: number }), bytes: body.length }
   }
 
   async call(rangeId: number, action: 'finish' | 'release' | 'heartbeat') {