Rebuild: edge:node --max-mbit caps a node's download

For workers on a shared line (poweredge-kantoor: 160 Mbit/s): the range
stream is read at most at the cap, TCP holds the coordinator back. The
cap is the node's, split over its processes (also EDGE_MAX_MBIT).
REBUILD.md: where workers run (connections set the speed; remote workers
take CPU off homelab).

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

REBUILD.md

@@ -45,6 +45,15 @@ search. Measured on the workstation (4 cores, 100 Mbit, Eweka):
   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.
 
+Where workers run: total speed is set by the account's NNTP connections
+(~30 k articles/s each), not by worker CPU, so remote workers don't make
+it faster; they take CPU off homelab (at 1.25 M/s ~7 worker cores) at the
+cost of relaying the compressed XOVER (≤ ~0.7 Gbit/s up, homelab has 1
+Gbit/s). Plan: coordinator, web, Postgres and a few workers on homelab,
+most workers on batcave, a few on poweredge-kantoor with
+`edge:node --max-mbit` (e.g. 50 of its 160 Mbit/s ≈ 100–125 k articles/s;
+the cap is shared by the node's processes).
+
 Not built yet: re-running ranges for a retracted rule (the counts and rule
 versions are recorded), deployment (Dockerfile, compose), phase 2.
 

commands/edge_node.ts

@@ -5,7 +5,7 @@ import { spawn, type ChildProcess } from 'node:child_process'
 import { availableParallelism } from 'node:os'
 import { promisify } from 'node:util'
 import { gzip, unzip } from 'node:zlib'
-import { RangeIndexer, decodeFrames, isKeepalive, toWire, type EdgeBatch, type Lease } from '#lib/edge/index'
+import { RangeIndexer, decodeFrames, isKeepalive, throttle, toWire, type EdgeBatch, type Lease } from '#lib/edge/index'
 
 const IDLE_MS = 30_000
 const unzipAsync = promisify(unzip)
@@ -37,6 +37,9 @@ export default class EdgeNode extends BaseCommand {
   @flags.number({ description: 'Worker processes (one core each; default: EDGE_PROCESSES or 1, 0 = one per core)' })
   declare processes?: number
 
+  @flags.number({ description: 'Download cap for the whole node in Mbit/s, shared by its processes (default: EDGE_MAX_MBIT or none)' })
+  declare maxMbit?: number
+
   private stopping = false
 
   async run() {
@@ -82,13 +85,21 @@ export default class EdgeNode extends BaseCommand {
    * (each gives its range back).
    */
   private async supervise(count: number) {
-    this.logger.info(`Starting ${count} worker processes.`)
+    const cap = this.cap()
+    this.logger.info(`Starting ${count} worker processes${cap ? ` (download cap ${cap} Mbit/s, ${(cap / count).toFixed(1)} each)` : ''}.`)
     const children = new Set<ChildProcess>()
     let stopping = false
     let stopped: () => void
     const done = new Promise<void>((resolve) => (stopped = resolve))
     const start = (slot: number) => {
-      const args = [process.argv[1], 'edge:node', `--batch-articles=${this.batchArticles}`, ...(this.ranges ? [`--ranges=${this.ranges}`] : [])]
+      const args = [
+        process.argv[1],
+        'edge:node',
+        `--batch-articles=${this.batchArticles}`,
+        ...(this.ranges ? [`--ranges=${this.ranges}`] : []),
+        // Each child gets its share of the node's cap.
+        ...(cap ? [`--max-mbit=${cap / count}`] : []),
+      ]
       const child = spawn(process.execPath, [...process.execArgv, ...args], { env: { ...process.env, EDGE_CHILD: String(slot) }, stdio: 'inherit' })
       children.add(child)
       child.on('exit', (code) => {
@@ -111,6 +122,12 @@ export default class EdgeNode extends BaseCommand {
     await this.terminate()
   }
 
+  /** The node's download cap in Mbit/s, or null. */
+  private cap(): number | null {
+    const mbit = this.maxMbit ?? (process.env.EDGE_MAX_MBIT ? Number(process.env.EDGE_MAX_MBIT) : null)
+    return mbit && mbit > 0 ? mbit : null
+  }
+
   private async runRange(api: Api, lease: Lease) {
     const { range, rules } = lease
     const started = Date.now()
@@ -157,7 +174,9 @@ export default class EdgeNode extends BaseCommand {
     try {
       // Inflating runs on a thread-pool thread: the next chunk inflates while this one is indexed.
       let pending: { from: number; to: number; text: Promise<Buffer> } | null = null
-      for await (const frame of decodeFrames(await api.stream(range.id))) {
+      const cap = this.cap()
+      const stream = await api.stream(range.id)
+      for await (const frame of decodeFrames(cap ? throttle(stream, (cap * 1e6) / 8) : stream)) {
         if (this.stopping) break
         if (isKeepalive(frame)) continue
         const next = { from: frame.from, to: frame.to, text: unzipAsync(frame.data) }

lib/edge/index.ts

@@ -4,3 +4,4 @@ export { toWire, fromWire, type WireBatch } from './wire.js'
 export type { Lease } from './lease.js'
 export { mergeFragments } from './merge.js'
 export type { EdgeSegment, FileMeta } from './scan.js'
+export { throttle } from './throttle.js'

lib/edge/throttle.ts

@@ -0,0 +1,18 @@
+import { setTimeout as sleep } from 'node:timers/promises'
+
+/**
+ * Passes a byte stream on at most `bytesPerSecond` on average: a worker on a
+ * line it shares (`edge:node --max-mbit`) reads its range slower, and TCP
+ * holds the sender back. The first second's worth goes at once.
+ */
+export async function* throttle<T extends Uint8Array>(stream: AsyncIterable<T> | Iterable<T>, bytesPerSecond: number): AsyncGenerator<T> {
+  const started = Date.now()
+  let bytes = 0
+  for await (const piece of stream) {
+    yield piece
+    bytes += piece.byteLength
+    const due = started + ((bytes - bytesPerSecond) / bytesPerSecond) * 1000
+    const wait = due - Date.now()
+    if (wait > 0) await sleep(wait)
+  }
+}

tests/unit/edge/throttle.spec.ts

@@ -0,0 +1,16 @@
+import { test } from '@japa/runner'
+import { throttle } from '#lib/edge/index'
+
+test.group('edge / throttle', () => {
+  test('passes everything on, at most the given rate after the first second', async ({ assert }) => {
+    const pieces = Array.from({ length: 30 }, (_, i) => new Uint8Array(1000).fill(i))
+    const started = Date.now()
+    const out: Uint8Array[] = []
+    // 30,000 bytes at 20,000 B/s: the first 20,000 at once, the rest over ~0.5 s.
+    for await (const piece of throttle(pieces, 20_000)) out.push(piece)
+    const seconds = (Date.now() - started) / 1000
+    assert.deepEqual(out, pieces)
+    assert.isAtLeast(seconds, 0.45)
+    assert.isBelow(seconds, 1.2)
+  })
+})