diff --git a/commands/edge_node.ts b/commands/edge_node.ts index 9150b52..f258122 100644 --- a/commands/edge_node.ts +++ b/commands/edge_node.ts @@ -15,8 +15,8 @@ class LeaseLost extends Error {} /** * An edge worker (REBUILD.md, part 2): leases a range from the coordinator, - * streams its chunks, indexes them, posts a batch every --batch-chunks - * chunks (each moves the range's resume point), and finishes the range. + * streams its chunks, indexes them, posts a batch every --batch-articles + * articles (each moves the range's resume point), and finishes the range. * Needs COORDINATOR_URL and NODE_TOKEN only. On SIGTERM it gives its range * back after the current chunk; the next worker resumes from the last batch. */ @@ -25,8 +25,8 @@ export default class EdgeNode extends BaseCommand { static description = 'Indexes ranges for the coordinator' static options: CommandOptions = { startApp: true, staysAlive: true } - @flags.number({ description: 'Chunks per posted batch (500k articles: ~4% of releases get a second chunk; LOG.md 2026-10-09)', default: 10 }) - declare batchChunks: number + @flags.number({ description: 'Articles per posted batch (500k: ~4% of releases get a second chunk; LOG.md 2026-10-09)', default: 500_000 }) + declare batchArticles: number @flags.number({ description: 'Stop after this many ranges (default: keep going)' }) declare ranges?: number @@ -85,7 +85,7 @@ export default class EdgeNode extends BaseCommand { let stopped: () => void const done = new Promise((resolve) => (stopped = resolve)) const start = (slot: number) => { - const args = [process.argv[1], 'edge:node', `--batch-chunks=${this.batchChunks}`, ...(this.ranges ? [`--ranges=${this.ranges}`] : [])] + const args = [process.argv[1], 'edge:node', `--batch-articles=${this.batchArticles}`, ...(this.ranges ? [`--ranges=${this.ranges}`] : [])] const child = spawn(process.execPath, [...process.execArgv, ...args], { env: { ...process.env, EDGE_CHILD: String(slot) }, stdio: 'inherit' }) children.add(child) child.on('exit', (code) => { @@ -113,7 +113,7 @@ export default class EdgeNode extends BaseCommand { const started = Date.now() const indexer = new RangeIndexer({ group: range.group, start: range.start, end: range.end, resumeAt: range.next, rules }) const totals = { headers: 0, releases: 0, candidates: 0, bytes: 0, postMs: 0 } - let chunks = 0 + let batchFrom = range.next let reachedEnd = false // Posts go in order, but the worker indexes on meanwhile (a post took 1–16 s; LOG.md 2026-10-09). let posting: Promise = Promise.resolve() @@ -136,10 +136,12 @@ export default class EdgeNode extends BaseCommand { const handle = async (chunk: { to: number; text: Promise }) => { if (postError) throw postError indexer.addText((await chunk.text).toString('utf8'), chunk.to + 1) - chunks++ reachedEnd = chunk.to >= range.end if (reachedEnd) indexer.finish() - if (reachedEnd || chunks % this.batchChunks === 0) post(indexer.take()) + if (reachedEnd || chunk.to + 1 - batchFrom >= this.batchArticles) { + post(indexer.take()) + batchFrom = chunk.to + 1 + } } try { // Inflating runs on a thread-pool thread: the next chunk inflates while this one is indexed. diff --git a/config/coordinator.ts b/config/coordinator.ts index 2492fe8..1030252 100644 --- a/config/coordinator.ts +++ b/config/coordinator.ts @@ -5,8 +5,13 @@ import app from '@adonisjs/core/services/app' export default { /** Articles per range. */ rangeArticles: env.get('RANGE_ARTICLES', 2_000_000), - /** Articles per XOVER chunk (spool files, stream frames). */ - chunkArticles: env.get('CHUNK_ARTICLES', 50_000), + /** + * Articles per XOVER chunk (spool files, stream frames). Eweka takes + * ~1.5–2 s per command whatever its size: one connection read 9–15k + * lines/s in chunks of 25k, ~20k in 50k, ~30k in 100k and 200k (LOG.md + * 2026-10-09). + */ + chunkArticles: env.get('CHUNK_ARTICLES', 100_000), /** A leased range without a heartbeat for this long goes back to the queue. */ leaseTtlSeconds: env.get('LEASE_TTL_SECONDS', 15 * 60), /** NNTP connections the coordinator fetches headers with. */