diff --git a/app/controllers/node_controller.ts b/app/controllers/node_controller.ts index af0c9bc..852a72d 100644 --- a/app/controllers/node_controller.ts +++ b/app/controllers/node_controller.ts @@ -74,6 +74,12 @@ export default class NodeController { return this.lifecycle(ctx, (rangeId, nodeId) => Leases.finish(rangeId, nodeId)) } + /** POST /ranges/:id/corrupt { from }: a chunk that doesn't inflate; the range goes back. */ + async corrupt(ctx: HttpContext) { + const from = Number(ctx.request.input('from')) + return this.lifecycle(ctx, (rangeId, nodeId) => Leases.corrupt(rangeId, nodeId, from)) + } + async release(ctx: HttpContext) { return this.lifecycle(ctx, (rangeId, nodeId) => Leases.release(rangeId, nodeId)) } diff --git a/app/services/leases.ts b/app/services/leases.ts index d558679..27e9407 100644 --- a/app/services/leases.ts +++ b/app/services/leases.ts @@ -58,6 +58,18 @@ class Leases { await Spool.remove(rangeId) } + /** + * A worker got a chunk that doesn't inflate: the spooled chunk goes and + * the range back to the queue (the prefetcher fetches leased ranges' + * missing chunks again; the next worker resumes at the range's resume point). + */ + async corrupt(rangeId: number, nodeId: number, from: number) { + const range = await this.leased(rangeId, nodeId) + const span = Spool.chunksOf({ start: Number(range.start_id), end: Number(range.end_id) }).find((s) => s.from === from) + if (span) await Spool.removeChunk(rangeId, span) + await db.from('ranges').where('id', rangeId).update({ status: 'planned', node_id: null }) + } + /** A stopping worker gives its range back; the next one resumes it. */ async release(rangeId: number, nodeId: number) { await this.leased(rangeId, nodeId) diff --git a/app/services/spool.ts b/app/services/spool.ts index 8b35292..2f21313 100644 --- a/app/services/spool.ts +++ b/app/services/spool.ts @@ -53,6 +53,11 @@ class Spool { return readFile(this.path(rangeId, span)) } + /** One chunk, e.g. one that doesn't inflate (a false end of reply; see XoverClient.compressedRaw). */ + async removeChunk(rangeId: number, span: ChunkSpan) { + await rm(this.path(rangeId, span), { force: true }) + } + async remove(rangeId: number) { await rm(join(this.dir, String(rangeId)), { recursive: true, force: true }) } diff --git a/commands/edge_node.ts b/commands/edge_node.ts index 5c45e0f..809cd9e 100644 --- a/commands/edge_node.ts +++ b/commands/edge_node.ts @@ -13,6 +13,8 @@ const gzipAsync = promisify(gzip) const RETRIES = 8 class LeaseLost extends Error {} +/** A spooled chunk that doesn't inflate (reported; the coordinator fetches it again). */ +class CorruptChunk extends Error {} /** * An edge worker (REBUILD.md, part 2): leases a range from the coordinator, @@ -134,9 +136,17 @@ export default class EdgeNode extends BaseCommand { posting.catch((error) => (postError ??= error)) } let postError: Error | null = null - const handle = async (chunk: { to: number; text: Promise }) => { + const handle = async (chunk: { from: number; to: number; text: Promise }) => { if (postError) throw postError - indexer.addText((await chunk.text).toString('utf8'), chunk.to + 1) + let text: Buffer + try { + text = await chunk.text + } catch (error: any) { + await posting.catch(() => {}) + await api.corrupt(range.id, chunk.from) + throw new CorruptChunk(`chunk ${chunk.from}–${chunk.to} doesn't inflate (${error.message}); reported, the range goes back`) + } + indexer.addText(text.toString('utf8'), chunk.to + 1) reachedEnd = chunk.to >= range.end if (reachedEnd) indexer.finish() if (reachedEnd || chunk.to + 1 - batchFrom >= this.batchArticles) { @@ -146,11 +156,11 @@ 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: { to: number; text: Promise } | null = null + let pending: { from: number; to: number; text: Promise } | 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) } + const next = { from: frame.from, to: frame.to, text: unzipAsync(frame.data) } next.text.catch(() => {}) if (pending) await handle(pending) pending = next @@ -172,7 +182,7 @@ export default class EdgeNode extends BaseCommand { ) } catch (error: any) { await posting.catch(() => {}) - if (error instanceof LeaseLost) { + if (error instanceof LeaseLost || error instanceof CorruptChunk) { this.logger.warning(`Range ${range.id}: ${error.message}`) return } @@ -227,6 +237,11 @@ class Api { return { ...((await response.json()) as { releases: number; candidates: number }), bytes: body.length } } + async corrupt(rangeId: number, from: number) { + const response = await this.request(`ranges/${rangeId}/corrupt`, { method: 'POST', body: JSON.stringify({ from }) }) + if (!response.ok) throw new Error(`corrupt: HTTP ${response.status}`) + } + async call(rangeId: number, action: 'finish' | 'release' | 'heartbeat') { const response = await this.request(`ranges/${rangeId}/${action}`, { method: 'POST', body: '{}' }) if (!response.ok) throw new Error(`${action}: HTTP ${response.status}`) diff --git a/lib/nntp/xover_client.ts b/lib/nntp/xover_client.ts index b004b89..eab8c69 100644 --- a/lib/nntp/xover_client.ts +++ b/lib/nntp/xover_client.ts @@ -233,20 +233,31 @@ export class XoverClient { } /** - * A zlib/gzip stream followed by a "." line: complete once it ends in - * ".\r\n" and inflates. The pieces are joined only then: joining them on - * every arrival was quadratic, ~15% of a prefetcher's time (LOG.md 2026-10-09). + * A zlib/gzip stream followed by a "." line. The pieces are joined only + * once the data ends in ".\r\n": joining them on every arrival was + * quadratic, ~15% of a prefetcher's time (LOG.md 2026-10-09). + * + * Raw (the coordinator's spool) it is complete then: inflating every reply + * to be sure cost the coordinator ~1.5 cores at ~630 k articles/s (mostly + * GC of the inflated text). ".\r\n" can also end a network read inside + * the compressed data (~1 in 16 M reads): that chunk doesn't inflate on + * the worker, which reports it and the chunk is fetched again (REBUILD.md, + * prefetch), and the rest of the reply breaks this connection's next + * command. With the text wanted, it must inflate. */ private async compressedRaw(wantText: boolean): Promise<{ raw: Buffer; text: string }> { while (true) { if (this.length > DOT_LINE.length && this.endsWith(DOT_LINE)) { const all = this.buffer() const raw = all.subarray(0, all.length - DOT_LINE.length) + if (!wantText) { + this.take(all.length) + return { raw, text: '' } + } try { - // 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') : '' } + return { raw, text: inflated.toString('utf8') } } catch { // ".\r\n" inside the compressed data: wait for more. } diff --git a/start/routes.ts b/start/routes.ts index e436b13..f0ad8a8 100644 --- a/start/routes.ts +++ b/start/routes.ts @@ -51,6 +51,7 @@ router router.post('ranges/:id/heartbeat', [controllers.Node, 'heartbeat']) router.post('ranges/:id/finish', [controllers.Node, 'finish']) router.post('ranges/:id/release', [controllers.Node, 'release']) + router.post('ranges/:id/corrupt', [controllers.Node, 'corrupt']) }) .prefix('node') .as('node')