Rebuild spec: measurements after the second performance round

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

REBUILD.md

@@ -14,17 +14,27 @@ prefetch spool, leases, rule sets, fragment merger, settling), the node API
 (`edge:node`, `--processes`), the catalog and chunk store, NZBs and a name
 search. Measured on the workstation (4 cores, 100 Mbit, Eweka):
 
-- Edge worker, one core: ~190–215 k headers/s on recent a.b.boneless,
+- Edge worker, one core: ~210–215 k headers/s on recent a.b.boneless,
-  ~330 k on a 2013 a.b.ath range; ~0.66 GB peak per process. (Start:
+  ~330–340 k on a 2013 a.b.ath range; ~0.7–0.8 GB peak per process.
-  73–103 k/s, 1.0–1.2 GB.)
+  (Start: 73–103 k/s, 1.0–1.2 GB.) What's left is spread out (the
-- NNTP: ~22–30 k overview lines/s per connection in chunks of 100 k
+  per-file key ~12%, GC ~10%, zlib ~10%, the scan ~6%).
-  (Eweka takes ~1.5–2 s per command); connections don't slow each other.
+- NNTP: Eweka takes ~1.5–2 s per XOVER command; ~22–30 k lines/s per
-- End to end (coordinator with 10 connections, web server, Postgres and 3
+  connection in chunks of 100 k when the line allows; connections don't
-  worker processes on one 4-core machine): 244 k headers/s; the server side
+  slow each other.
-  used 10% (web) + 22% (coordinator) + 32% (Postgres) of a core.
+- Bandwidth: compressed XOVER is ~69 B per article on recent boneless
+  (100 k-article chunks of ~7 MB). The workstation's 100 Mbit line carries
+  ~140–150 k articles/s, which is where its end-to-end runs now stop (10
+  connections 90–99% busy, ~89 Mbit/s).
+- End to end at that rate (one 4-core machine): workers about half busy
+  each; server side ~17% (coordinator) + ~8% (web) + ~3% (Postgres) of a
+  core. (Before this round's fixes Postgres was at 32%: the merger's
+  neighbour query read every range per file.)
 - Storage: chunks 17.9–20.3 B per segment (format v2); a release ~1.1 KB.
-- For the whole target (528 B overview lines): ~4–6 days at 50 connections,
+- For the whole target (528 B overview lines): ~36 TB of compressed XOVER
-  ~3 worker cores.
+  to download. At 50 connections (~1–1.25 M lines/s, ~0.6–0.7 Gbit/s)
+  ~5–6 days, ~5–6 worker cores, two coordinator shards (one process
+  handles ~60 MB/s), under a core for web and Postgres. The line is the
+  limit below ~0.7 Gbit/s: at 1 Gbit, ~3.5 days at best.
 
 Not built yet: re-running ranges for a retracted rule (the counts and rule
 versions are recorded), deployment (Dockerfile, compose), phase 2.
@@ -46,14 +56,14 @@ dropping garbage; 28 TB of compressed overview traffic. Throughput goal:
    account; the gzip stream is passed on as is).
 
    **Prefetch**: the coordinator fetches the next ranges ahead of the
-   workers, in chunks of 50 k lines and in order, into a spool on disk
+   workers, in chunks of 100 k lines and in order, into a spool on disk
    (~106 MB per 2 M range at ~53 B per line; on the HDD, written and read
    sequentially). It keeps about two ranges ahead per active lease, capped
    by size (default 10 GB), and fetches with all its header connections all
    the time, independent of how fast workers are. A lease goes to a range
-   that is already (partly) spooled; a worker reads its chunks from the
+   that is already (partly) spooled, else to the next planned one (leased
-   spool, waiting only for chunks still being fetched, and falls back to a
+   ranges are fetched first); a worker reads its chunks from the spool,
-   direct relay fetch for a range that isn't spooled. A worker gets its
+   waiting only for chunks still being fetched. A worker gets its
    whole range in one streaming request (`GET /ranges/:id/stream`): the
    spooled chunks in order, waiting for those still being fetched. It still
    parses chunk by chunk (2 M lines are ~560 MB of text) and posts results
@@ -64,6 +74,11 @@ dropping garbage; 28 TB of compressed overview traffic. Throughput goal:
    range is leased again later.
    Order: oldest era first (2008–2018 holds 63% of the readable files on
    17% of the lines), then forward.
+   One coordinator process handles ~60 MB/s of compressed XOVER (its event
+   loop was 16.5% busy at ~10 MB/s); beyond that, prefetching is split over
+   processes with `coordinator:run --shard=i/n` (ranges with id % n = i − 1,
+   each with its own NNTP_CONNECTIONS; only shard 1 plans, merges and
+   settles).
 
 2. **Edge worker** (node, TypeScript reusing `lib/`). Measured (`edge:bench`,
    2026-10-09): 73 k headers/s per core on an a.b.ath range from 2013,

app/services/fragment_merger.ts

@@ -120,17 +120,22 @@ class FragmentMerger {
     })
     const spans = lists.map(span)
     const groups = [...new Set(spans.map((s) => s.group))]
+    // The range holding the article before and the one after, each by the (group_id, start_id) index:
+    // "end_id = ? OR start_id = ?" read every range of the group per file (5.5 s a round; LOG.md 2026-10-09).
     const { rows } = await db.rawQuery(
-      `SELECT r.group_id, r.start_id, r.end_id, r.status FROM ranges r
+      `SELECT s.group_id, s.before_end, s.after_start, b.status AS before, a.status AS after
-       JOIN unnest(?::int[], ?::bigint[], ?::bigint[]) AS s(group_id, before_end, after_start)
+       FROM unnest(?::int[], ?::bigint[], ?::bigint[]) AS s(group_id, before_end, after_start)
-         ON r.group_id = s.group_id AND (r.end_id = s.before_end OR r.start_id = s.after_start)`,
+       LEFT JOIN LATERAL (SELECT r.status FROM ranges r WHERE r.group_id = s.group_id AND r.start_id <= s.before_end AND r.end_id >= s.before_end
+                          ORDER BY r.start_id DESC LIMIT 1) b ON true
+       LEFT JOIN LATERAL (SELECT r.status FROM ranges r WHERE r.group_id = s.group_id AND r.start_id <= s.after_start AND r.end_id >= s.after_start
+                          ORDER BY r.start_id DESC LIMIT 1) a ON true`,
       [spans.map((s) => s.group), spans.map((s) => s.first - 1), spans.map((s) => s.last + 1)]
     )
     const byEnd = new Map<string, string>()
     const byStart = new Map<string, string>()
     for (const r of rows) {
-      byEnd.set(`${r.group_id}:${r.end_id}`, r.status)
+      if (r.before !== null) byEnd.set(`${r.group_id}:${r.before_end}`, r.before)
-      byStart.set(`${r.group_id}:${r.start_id}`, r.status)
+      if (r.after !== null) byStart.set(`${r.group_id}:${r.after_start}`, r.after)
     }
     const lows = new Map((await db.from('groups').whereIn('id', groups).select('id', 'low')).map((g) => [Number(g.id), g.low === null ? null : Number(g.low)]))
     return (list: (typeof lists)[number]) => {

app/services/leases.ts

@@ -7,23 +7,30 @@ import type { Lease } from '#lib/edge/index'
 export class LeaseLost extends Error {}
 
 /**
- * Ranges for workers (REBUILD.md, coordinator): the next range whose first
+ * Ranges for workers (REBUILD.md, coordinator): the next range with chunks
- * chunk is spooled, by priority (oldest era first); a lease whose worker
+ * spooled, by priority (oldest era first), else the next planned one (the
- * went quiet for LEASE_TTL_SECONDS is taken over and resumes at its resume
+ * prefetcher fetches leased ranges first; the worker's stream waits). A
- * point.
+ * worker never waits for prefetching to choose its range: refused leases
+ * left workers idle for 30 s, so fewer ranges were leased and fewer
+ * prefetched (LOG.md 2026-10-09). A lease whose worker went quiet for
+ * LEASE_TTL_SECONDS is taken over and resumes at its resume point.
  */
 class Leases {
   async lease(nodeId: number): Promise<Lease | null> {
     const rules = await RuleSets.current()
-    const { rows } = await db.rawQuery(
+    const take = (spooled: boolean) =>
-      `UPDATE ranges r SET status = 'leased', node_id = ?, leased_at = now(), heartbeat_at = now(), attempts = attempts + 1, rule_version = ?
+      db.rawQuery(
-       WHERE r.id = (
+        `UPDATE ranges r SET status = 'leased', node_id = ?, leased_at = now(), heartbeat_at = now(), attempts = attempts + 1, rule_version = ?
-         SELECT id FROM ranges
+         WHERE r.id = (
-         WHERE spooled_chunks > 0 AND (status = 'planned' OR (status = 'leased' AND heartbeat_at < now() - make_interval(secs => ?)))
+           SELECT id FROM ranges
-         ORDER BY priority, id LIMIT 1 FOR UPDATE SKIP LOCKED)
+           WHERE ${spooled ? 'spooled_chunks > 0' : 'true'} AND next_id <= end_id
-       RETURNING r.id, r.group_id, r.start_id, r.end_id, r.next_id`,
+             AND (status = 'planned' OR (status = 'leased' AND heartbeat_at < now() - make_interval(secs => ?)))
-      [nodeId, rules.version, coordinatorConfig.leaseTtlSeconds]
+           ORDER BY priority, id LIMIT 1 FOR UPDATE SKIP LOCKED)
-    )
+         RETURNING r.id, r.group_id, r.start_id, r.end_id, r.next_id`,
+        [nodeId, rules.version, coordinatorConfig.leaseTtlSeconds]
+      )
+    let { rows } = await take(true)
+    if (rows.length === 0) ({ rows } = await take(false))
     const row = rows[0]
     if (!row) return null
     await db.from('nodes').where('id', nodeId).update({ last_seen_at: new Date() })

app/services/nntp_pool.ts

@@ -10,6 +10,7 @@ import { XoverClient } from '#lib/nntp/index'
  */
 class NntpPool {
   private idle: XoverClient[] = []
+  private busy = new Set<XoverClient>()
   private open = 0
   private waiting: ((client: XoverClient) => void)[] = []
 
@@ -19,6 +20,7 @@ class NntpPool {
 
   async use<T>(work: (client: XoverClient) => Promise<T>): Promise<T> {
     const client = await this.acquire()
+    this.busy.add(client)
     let broken = false
     try {
       return await work(client)
@@ -26,6 +28,7 @@ class NntpPool {
       broken = true
       throw error
     } finally {
+      this.busy.delete(client)
       this.release(client, broken)
     }
   }
@@ -71,8 +74,10 @@ class NntpPool {
     else this.idle.push(client)
   }
 
+  /** Closes every connection, also those in use: their commands fail. */
   async close() {
-    for (const client of this.idle.splice(0)) client.close()
+    for (const client of [...this.idle.splice(0), ...this.busy]) client.close()
+    this.busy.clear()
     this.open = 0
   }
 }

app/services/prefetcher.ts

@@ -28,6 +28,13 @@ const SIZE_MS = 30_000
  * kept its worker waiting (LOG.md 2026-10-09).
  */
 export class Prefetcher {
+  /**
+   * This process's share of the ranges, `id % count = index` (`coordinator:run --shard`): one
+   * process's event loop handles ~60 MB/s of compressed XOVER (16.5% of a core at ~10 MB/s;
+   * LOG.md 2026-10-09), less than 50 connections bring.
+   */
+  constructor(private shard = { index: 0, count: 1 }) {}
+
   private targets: Target[] = []
   private claimed = new Set<string>()
   private running = 0
@@ -35,6 +42,8 @@ export class Prefetcher {
   private stopped = false
   private spoolBytes = 0
   private sizedAt = 0
+  /** Fetched since the last report: chunks, articles, ms spent fetching. */
+  private stats = { chunks: 0, articles: 0, fetchMs: 0, since: Date.now() }
 
   /** One round: refreshes the targets now and then, and puts idle connections to work. */
   async tick() {
@@ -42,6 +51,7 @@ export class Prefetcher {
       this.refreshedAt = Date.now()
       this.targets = await this.readTargets()
     }
+    this.report()
     while (!this.stopped && this.running < NntpPool.size && this.peek()) {
       this.running++
       this.work()
@@ -65,10 +75,14 @@ export class Prefetcher {
       if (!next) return
       const { target, span } = next
       try {
+        const fetchedAt = Date.now()
         const data = await NntpPool.use(async (client) => {
           await client.group(target.group)
           return client.xoverRaw(span.from, span.to)
         })
+        this.stats.chunks++
+        this.stats.articles += span.to - span.from + 1
+        this.stats.fetchMs += Date.now() - fetchedAt
         const existed = await Spool.has(target.id, span)
         await Spool.write(target.id, span, data)
         if (existed) continue
@@ -84,6 +98,20 @@ export class Prefetcher {
     }
   }
 
+  /** A line a minute: what the connections fetched, and how long a chunk took. */
+  private report() {
+    const seconds = (Date.now() - this.stats.since) / 1000
+    if (seconds < 60) return
+    const { chunks, articles, fetchMs } = this.stats
+    if (chunks > 0) {
+      logger.info(
+        `Prefetched ${chunks} chunks in ${seconds.toFixed(0)} s: ${Math.round(articles / seconds).toLocaleString('en')} articles/s, ` +
+          `${(fetchMs / chunks / 1000).toFixed(1)} s a chunk, connections busy ${Math.round((fetchMs / (seconds * 1000 * NntpPool.size)) * 100)}%.`
+      )
+    }
+    this.stats = { chunks: 0, articles: 0, fetchMs: 0, since: Date.now() }
+  }
+
   private peek(): boolean {
     return this.targets.some((t) => t.missing.some((s) => !this.claimed.has(`${t.id}:${s.from}`)))
   }
@@ -105,16 +133,17 @@ export class Prefetcher {
       this.sizedAt = Date.now()
       this.spoolBytes = await Spool.size()
     }
-    const rows = (q: any) => 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 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'))
     // 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.
     const ahead = Math.max(coordinatorConfig.spool.minAhead, coordinatorConfig.spool.aheadPerLease * leased.length)
     const started = await rows(db.query()).where('r.status', 'planned').where('r.spooled_chunks', '>', 0).whereNull('r.spooled_at').orderBy([{ column: 'r.priority' }, { column: 'r.id' }])
-    const [{ ready }] = await db.from('ranges').where('status', 'planned').where('spooled_chunks', '>', 0).count('* as ready')
+    const [{ ready }] = await mine(db.from('ranges as r').where('r.status', 'planned').where('r.spooled_chunks', '>', 0)).count('* as ready')
     const room = this.spoolBytes < coordinatorConfig.spool.maxBytes
     const fresh = room && Number(ready) < ahead
-      ? await rows(db.query()).where('r.status', 'planned').where('r.spooled_chunks', 0).orderBy([{ column: 'r.priority' }, { column: 'r.id' }]).limit(ahead - Number(ready))
+      ? await rows(db.query()).where('r.status', 'planned').where('r.spooled_chunks', 0).whereRaw('r.next_id <= r.end_id').orderBy([{ column: 'r.priority' }, { column: 'r.id' }]).limit(ahead - Number(ready))
       : []
     const targets: Target[] = []
     for (const row of [...leased, ...(room ? started : []), ...fresh]) {

commands/coordinator_run.ts

@@ -1,4 +1,4 @@
-import { BaseCommand } from '@adonisjs/core/ace'
+import { BaseCommand, flags } from '@adonisjs/core/ace'
 import type { CommandOptions } from '@adonisjs/core/types/ace'
 import { setTimeout as sleep } from 'node:timers/promises'
 import { Prefetcher } from '#services/prefetcher'
@@ -11,6 +11,7 @@ const TICK_MS = 2000
 const PLAN_EVERY_MS = 60 * 60 * 1000
 const MERGE_EVERY_MS = 60 * 1000
 const MERGE_BUDGET_MS = 30 * 1000
+const STOP_WAIT_MS = 15 * 1000
 
 /**
  * The coordinator's background work (REBUILD.md, part 1): the prefetcher,
@@ -23,30 +24,49 @@ export default class CoordinatorRun extends BaseCommand {
   static description = 'Prefetches ranges into the spool and plans new ones'
   static options: CommandOptions = { startApp: true, staysAlive: true }
 
+  @flags.string({ description: 'This process\'s share of the prefetching, "i/n" (ranges with id % n = i - 1); only 1/n plans, merges and settles' })
+  declare shard?: string
+
   async run() {
-    const prefetcher = new Prefetcher()
+    const [index, count] = (this.shard ?? '1/1').split('/').map(Number)
+    if (!(count >= 1 && index >= 1 && index <= count)) throw new Error(`--shard: "i/n" with 1 <= i <= n, not "${this.shard}"`)
+    const prefetcher = new Prefetcher({ index: index - 1, count })
+    const main = index === 1
     let stopping = false
     let stopped: () => void
     const done = new Promise<void>((resolve) => (stopped = resolve))
     // bin/console turns SIGTERM into app.terminate: wait for the spooling in flight first.
     this.app.terminating(async () => {
+      this.logger.info('Stopping: waiting for the chunks in flight.')
       stopping = true
       prefetcher.stop()
       await done
     })
+    this.logger.info(count > 1 ? `Coordinator started (shard ${index}/${count}${main ? '' : ', prefetching only'}).` : 'Coordinator started.')
+    // Two loops: the prefetcher's ticks don't wait for merging or planning (a merge round
+    // of up to MERGE_BUDGET_MS left the connections idle once their targets were fetched).
+    const prefetching = (async () => {
+      while (!stopping) {
+        try {
+          await prefetcher.tick()
+        } catch (error: any) {
+          this.logger.error(`Prefetch round failed: ${error.message}`)
+        }
+        await sleep(TICK_MS)
+      }
+    })()
     let plannedAt = 0
     let mergedAt = 0
-    this.logger.info('Coordinator started.')
     while (!stopping) {
       try {
-        if (Date.now() - plannedAt >= PLAN_EVERY_MS) {
+        if (main && Date.now() - plannedAt >= PLAN_EVERY_MS) {
           plannedAt = Date.now()
           const { groups, ranges } = await Planner.planAll()
           if (ranges > 0) this.logger.info(`Planned ${ranges} ranges over ${groups} groups.`)
           const settled = await Settler.run()
           if (settled > 0) this.logger.info(`Settled ${settled} releases.`)
         }
-        if (Date.now() - mergedAt >= MERGE_EVERY_MS) {
+        if (main && Date.now() - mergedAt >= MERGE_EVERY_MS) {
           mergedAt = Date.now()
           // Rounds until a whole pass over the file keys decides nothing more, or MERGE_BUDGET_MS.
           // (Going on while only waiting files were left kept Postgres busy all the time.)
@@ -65,14 +85,19 @@ export default class CoordinatorRun extends BaseCommand {
           }
           if (totals.completed || totals.incomplete) this.logger.info(`Fragments: ${totals.completed} files completed, ${totals.incomplete} incomplete, ${totals.waiting} waiting.`)
         }
-        await prefetcher.tick()
       } catch (error: any) {
         this.logger.error(`Coordinator round failed: ${error.message}`)
       }
       await sleep(TICK_MS)
     }
-    while (prefetcher.active > 0) await sleep(200)
+    await prefetching
+    // The chunks in flight, for at most STOP_WAIT_MS (a connection the server holds back can take minutes); then closed.
+    const waitUntil = Date.now() + STOP_WAIT_MS
+    while (prefetcher.active > 0 && Date.now() < waitUntil) await sleep(200)
+    if (prefetcher.active > 0) this.logger.info(`Closing ${prefetcher.active} connections still fetching.`)
     await NntpPool.close()
+    this.logger.info('Coordinator stopped.')
     stopped!()
+    await this.terminate()
   }
 }

commands/edge_bench.ts

@@ -163,6 +163,8 @@ export default class EdgeBench extends BaseCommand {
         peakDetail = `heap ${(m.heapUsed / 1e6).toFixed(0)}/${(m.heapTotal / 1e6).toFixed(0)} MB, external ${(m.external / 1e6).toFixed(0)} MB, array buffers ${(m.arrayBuffers / 1e6).toFixed(0)} MB`
       }
     }
+    // Done, so nothing takes it for a range to work on (a leased one was set back to planned once).
+    if (this.store) await db.from('ranges').where('id', rangeId).update({ status: 'done', done_at: new Date() })
     const cpu = process.cpuUsage(cpuStart)
     const cpuS = (cpu.user + cpu.system) / 1e6
     const total = t.inflate + t.parse + t.index

commands/edge_node.ts

@@ -4,11 +4,12 @@ import { setTimeout as sleep } from 'node:timers/promises'
 import { spawn, type ChildProcess } from 'node:child_process'
 import { availableParallelism } from 'node:os'
 import { promisify } from 'node:util'
-import { gzipSync, unzip } from 'node:zlib'
+import { gzip, unzip } from 'node:zlib'
 import { RangeIndexer, decodeFrames, isKeepalive, toWire, type EdgeBatch, type Lease } from '#lib/edge/index'
 
 const IDLE_MS = 30_000
 const unzipAsync = promisify(unzip)
+const gzipAsync = promisify(gzip)
 const RETRIES = 8
 
 class LeaseLost extends Error {}
@@ -147,12 +148,12 @@ export default class EdgeNode extends BaseCommand {
       // Inflating runs on a thread-pool thread: the next chunk inflates while this one is indexed.
       let pending: { to: number; text: Promise<Buffer> } | null = null
       for await (const frame of decodeFrames(await api.stream(range.id))) {
+        if (this.stopping) break
         if (isKeepalive(frame)) continue
         const next = { to: frame.to, text: unzipAsync(frame.data) }
         next.text.catch(() => {})
         if (pending) await handle(pending)
         pending = next
-        if (this.stopping) break
       }
       if (pending && !this.stopping) await handle(pending)
       await posting
@@ -219,8 +220,8 @@ class Api {
   }
 
   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.
+    // Gzipped (on a thread-pool thread): chunk payloads are compressed already, the rest compresses well.
-    const body = gzipSync(JSON.stringify(toWire(batch)), { level: 1 })
+    const body = await gzipAsync(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 { ...((await response.json()) as { releases: number; candidates: number }), bytes: body.length }

database/migrations/1770000001100_ranges_prefetch_indexes.ts

@@ -0,0 +1,18 @@
+import { BaseSchema } from '@adonisjs/lucid/schema'
+
+/**
+ * The prefetcher's ranges every 2 s (REBUILD.md, prefetch): the leased ones
+ * and the planned ones it started spooling. Without these it read all
+ * ranges each time (~15–25 ms with 145k planned; LOG.md 2026-10-09).
+ */
+export default class extends BaseSchema {
+  async up() {
+    this.schema.raw(`CREATE INDEX IF NOT EXISTS ranges_leased_idx ON ranges (leased_at) WHERE status = 'leased'`)
+    this.schema.raw(`CREATE INDEX IF NOT EXISTS ranges_spooling_idx ON ranges (priority, id) WHERE status = 'planned' AND spooled_chunks > 0`)
+  }
+
+  async down() {
+    this.schema.raw(`DROP INDEX IF EXISTS ranges_spooling_idx`)
+    this.schema.raw(`DROP INDEX IF EXISTS ranges_leased_idx`)
+  }
+}

lib/edge/range_indexer.ts

@@ -1,4 +1,4 @@
-import { createHash } from 'node:crypto'
+import { hash } from 'node:crypto'
 import type { Overview } from '../nntp/index.js'
 import { FileAggregator, type AggregatedFile, isArchiveStart, isLaterVolume, sevenZipVolume } from '../files/index.js'
 import { RuleSet, isRandomFile, uploadTool, type Rules } from '../rules/index.js'
@@ -281,7 +281,7 @@ export class RangeIndexer {
 
 /** The release key; a file without a recognisable release becomes a release of its own. */
 export function releaseKeyOf(s: EdgeSegment): string {
-  return s.file.release?.key ?? createHash('sha1').update(`file\n${s.fileKey}`).digest('hex')
+  return s.file.release?.key ?? hash('sha1', `file\n${s.fileKey}`)
 }
 
 function postedOf(segments: EdgeSegment[]): number | null {

lib/edge/stream.ts

@@ -26,17 +26,33 @@ export function encodeFrame(frame: Frame): Buffer {
   return Buffer.concat([head, frame.data])
 }
 
-/** The frames of a byte stream, as they complete. */
+/**
+ * The frames of a byte stream, as they complete. Pieces are kept in a list
+ * and joined once a frame is complete: joining each piece onto what came
+ * before copied a chunk of MBs once per 64 KB (~5% of a worker's time;
+ * LOG.md 2026-10-09).
+ */
 export async function* decodeFrames(stream: AsyncIterable<Uint8Array> | Iterable<Uint8Array>): AsyncGenerator<Frame> {
-  let buffer = Buffer.alloc(0)
+  let pieces: Buffer[] = []
+  let length = 0
+  /** Bytes needed before the next frame can be cut: its header first, then all of it. */
+  let needed = HEADER
   for await (const piece of stream) {
-    buffer = buffer.length === 0 ? Buffer.from(piece) : Buffer.concat([buffer, piece])
+    pieces.push(Buffer.isBuffer(piece) ? piece : Buffer.from(piece.buffer, piece.byteOffset, piece.byteLength))
-    while (buffer.length >= HEADER) {
+    length += piece.byteLength
-      const length = buffer.readUInt32BE(16)
+    while (length >= needed) {
-      if (buffer.length < HEADER + length) break
+      let buffer = pieces.length === 1 ? pieces[0] : Buffer.concat(pieces, length)
-      yield { from: Number(buffer.readBigUInt64BE(0)), to: Number(buffer.readBigUInt64BE(8)), data: buffer.subarray(HEADER, HEADER + length) }
+      pieces = [buffer]
-      buffer = buffer.subarray(HEADER + length)
+      if (needed === HEADER) {
+        needed = HEADER + buffer.readUInt32BE(16)
+        if (length < needed) break
+      }
+      yield { from: Number(buffer.readBigUInt64BE(0)), to: Number(buffer.readBigUInt64BE(8)), data: buffer.subarray(HEADER, needed) }
+      buffer = buffer.subarray(needed)
+      pieces = buffer.length > 0 ? [buffer] : []
+      length = buffer.length
+      needed = HEADER
     }
   }
-  if (buffer.length > 0) throw new Error(`Stream ended inside a frame (${buffer.length} bytes left)`)
+  if (length > 0) throw new Error(`Stream ended inside a frame (${length} bytes left)`)
 }

lib/headers/file_key.ts

@@ -1,4 +1,4 @@
-import { createHash } from 'node:crypto'
+import { hash } from 'node:crypto'
 
 /**
  * Identity of one uploaded file: every segment of it has the same poster, the
@@ -7,7 +7,7 @@ import { createHash } from 'node:crypto'
  * See CONCEPT.md (D12).
  */
 export function fileKey(poster: string, base: string, totalSegments: number): string {
-  return createHash('sha1').update(`${poster}\n${base}\n${totalSegments}`).digest('hex')
+  return hash('sha1', `${poster}\n${base}\n${totalSegments}`)
 }
 
 /** True when a yEnc name looks like a real filename (it has an extension), not a random token. */
@@ -26,7 +26,7 @@ export function looksLikeFilename(name: string): boolean {
  */
 export function weakFileKey(name: string, size: number, totalParts: number): string {
   const identity = looksLikeFilename(name) ? `name\n${name}` : 'shape'
-  return createHash('sha1').update(`weak\n${identity}\n${size}\n${totalParts}`).digest('hex')
+  return hash('sha1', `weak\n${identity}\n${size}\n${totalParts}`)
 }
 
 /**
@@ -41,7 +41,5 @@ export function counterFileKey(segment: { poster: string; subject: string; total
   const counter = /\[(\d+)\/(\d+)\]/.exec(segment.subject)
   if (!counter || segment.totalParts < 2) return null
   const day = segment.postedAt && !Number.isNaN(segment.postedAt.getTime()) ? segment.postedAt.toISOString().slice(0, 10) : ''
-  return createHash('sha1')
+  return hash('sha1', `counter\n${segment.poster}\n${Number(counter[1])}/${Number(counter[2])}\n${segment.totalParts}\n${day}`)
-    .update(`counter\n${segment.poster}\n${Number(counter[1])}/${Number(counter[2])}\n${segment.totalParts}\n${day}`)
-    .digest('hex')
 }

lib/nntp/xover_client.ts

@@ -1,9 +1,10 @@
 import { promisify } from 'node:util'
 import net from 'node:net'
 import tls from 'node:tls'
-import { deflate, deflateSync, unzipSync } from 'node:zlib'
+import { deflate, deflateSync, unzip } from 'node:zlib'
 
 const deflateAsync = promisify(deflate)
+const unzipAsync = promisify(unzip)
 
 export interface XoverClientOptions {
   host: string
@@ -66,7 +67,8 @@ export class XoverClient {
       ? 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)
-    sock.setTimeout(client.timeoutMs)
+    // The greeting and login are short: a server holding a connection back (too many open) fails it in 30 s, not timeoutMs.
+    sock.setTimeout(Math.min(client.timeoutMs, 30_000))
     try {
       await client.expect(await client.line(), ['200', '201'])
       if (options.user) {
@@ -241,7 +243,8 @@ export class XoverClient {
         const all = this.buffer()
         const raw = all.subarray(0, all.length - DOT_LINE.length)
         try {
-          const inflated = unzipSync(raw)
+          // On a thread-pool thread: a chunk's ~25 MB of text took the coordinator's event loop ~60 ms.
+          const inflated = await unzipAsync(raw)
           this.take(all.length)
           return { raw: Buffer.from(raw), text: wantText ? inflated.toString('utf8') : '' }
         } catch {

lib/release/identity.ts

@@ -1,4 +1,4 @@
-import { createHash } from 'node:crypto'
+import { hash as digest } from 'node:crypto'
 import { looksLikeFilename } from '../headers/file_key.js'
 
 export interface ReleaseIdentity {
@@ -103,7 +103,7 @@ export function namesRelease(remainder: string): boolean {
 }
 
 function hash(...parts: (string | number | undefined)[]) {
-  return createHash('sha1').update(parts.map((p) => p ?? '').join('\n')).digest('hex')
+  return digest('sha1', parts.map((p) => p ?? '').join('\n'))
 }
 
 /**

lib/rules/upload_tools.ts

@@ -1,4 +1,4 @@
-import { createHash } from 'node:crypto'
+import { hash } from 'node:crypto'
 
 /**
  * Upload signatures (CONCEPT.md D42): which tool posted an article, read from
@@ -48,5 +48,5 @@ export function signatureKey(tool: string, group: string) {
  */
 export function explored(key: string, percent: number): boolean {
   if (percent <= 0) return false
-  return createHash('md5').update(key).digest().readUInt32BE(0) % 10_000 < percent * 100
+  return hash('md5', key, 'buffer').readUInt32BE(0) % 10_000 < percent * 100
 }

tests/unit/edge/stream.spec.ts

@@ -25,4 +25,15 @@ test.group('edge / stream frames', () => {
       }
     }, /inside a frame/)
   })
+  test('frames come out whole whatever the piece sizes, also pieces holding several frames', async ({ assert }) => {
+    const sent = Array.from({ length: 12 }, (_, i) => ({ from: i * 10 + 1, to: i * 10 + 10, data: Buffer.alloc((i * 997) % 5000, i) }))
+    const whole = Buffer.concat(sent.map(encodeFrame))
+    for (const size of [1, 3, 20, 21, 64, 1000, 4096, whole.length]) {
+      const pieces = []
+      for (let at = 0; at < whole.length; at += size) pieces.push(new Uint8Array(whole.subarray(at, at + size)))
+      const frames = []
+      for await (const frame of decodeFrames(pieces)) frames.push(frame)
+      assert.deepEqual(frames.map((f) => [f.from, f.to, f.data.toString('hex')]), sent.map((f) => [f.from, f.to, f.data.toString('hex')]), `pieces of ${size}`)
+    }
+  })
 })