From 42b6cac3ee5835de42137901d0ae112ea71cd86e Mon Sep 17 00:00:00 2001 From: Daan Meijer Date: Fri, 9 Oct 2026 09:01:57 +0200 Subject: [PATCH] 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 --- REBUILD.md | 9 +++++++++ commands/edge_node.ts | 27 +++++++++++++++++++++++---- lib/edge/index.ts | 1 + lib/edge/throttle.ts | 18 ++++++++++++++++++ tests/unit/edge/throttle.spec.ts | 16 ++++++++++++++++ 5 files changed, 67 insertions(+), 4 deletions(-) create mode 100644 lib/edge/throttle.ts create mode 100644 tests/unit/edge/throttle.spec.ts diff --git a/REBUILD.md b/REBUILD.md index ee8b595..730e40c 100644 --- a/REBUILD.md +++ b/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. diff --git a/commands/edge_node.ts b/commands/edge_node.ts index 809cd9e..5040534 100644 --- a/commands/edge_node.ts +++ b/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() let stopping = false let stopped: () => void const done = new Promise((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 } | 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) } diff --git a/lib/edge/index.ts b/lib/edge/index.ts index 8663d2a..b08668d 100644 --- a/lib/edge/index.ts +++ b/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' diff --git a/lib/edge/throttle.ts b/lib/edge/throttle.ts new file mode 100644 index 0000000..ddb71ad --- /dev/null +++ b/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(stream: AsyncIterable | Iterable, bytesPerSecond: number): AsyncGenerator { + 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) + } +} diff --git a/tests/unit/edge/throttle.spec.ts b/tests/unit/edge/throttle.spec.ts new file mode 100644 index 0000000..05d4df4 --- /dev/null +++ b/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) + }) +})