Rebuild: XOVER chunks of 100k lines; batches by articles
Eweka takes ~1.5–2 s per XOVER command whatever its size. One connection on recent a.b.boneless: 9–15k lines/s in chunks of 25k, ~20k in 50k, ~30k in 100k–200k. Four connections at once: 15k each at 50k, 22k each at 100k (the same as one connection alone at that moment): connections don't slow each other. CHUNK_ARTICLES defaults to 100k; edge:node posts a batch every --batch-articles (500k) instead of every N chunks. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
1 parent
08c876b690
commit
5ba93f558c
2 files changed
+17
-10
No files matched your search
+10
-8
@@ -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<void>((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<void> = Promise.resolve()
|
||||
@@ -136,10 +136,12 @@ export default class EdgeNode extends BaseCommand {
|
||||
const handle = async (chunk: { to: number; text: Promise<Buffer> }) => {
|
||||
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.
|
||||
|
||||
@@ -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. */
|
||||
|
||||
Reference in new issue
Block a user