Daan MeijerandClaude Opus 5.5
a979b5d11c
On homelab (1 Gbit, 20 connections, ~630k articles/s) the coordinator used 1.3-1.7 cores, mostly GC of the text it inflated only to know a compressed reply was complete, plus a copy of every reply. A raw reply is now complete when it ends in ".\r\n" (no copy): 0.4-0.5 cores at the same rate. ".\r\n" can also end a network read inside the compressed data (~1 in 16M reads, a few dozen times over the whole target). Such a chunk doesn't inflate on the worker: it reports it (POST ranges/:id/corrupt), the spooled chunk is removed and the range goes back to the queue; the prefetcher fetches it again. The rest of that reply breaks the connection's next command, so the connection is replaced. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
109 lines
4.0 KiB
TypeScript
109 lines
4.0 KiB
TypeScript
import type { HttpContext } from '@adonisjs/core/http'
|
||
import { Readable } from 'node:stream'
|
||
import { setTimeout as sleep } from 'node:timers/promises'
|
||
import Leases, { LeaseLost } from '#services/leases'
|
||
import BatchStore, { StaleBatch } from '#services/batch_store'
|
||
import Spool from '#services/spool'
|
||
import { KEEPALIVE, encodeFrame, fromWire, type WireBatch } from '#lib/edge/index'
|
||
|
||
/** How long the stream waits for a chunk the prefetcher hasn't spooled yet. */
|
||
const CHUNK_WAIT_MS = 15 * 60 * 1000
|
||
/** While waiting, an empty frame this often: an HTTP client gives up on a quiet body (Node's after 5 minutes). */
|
||
const KEEPALIVE_MS = 20_000
|
||
|
||
/**
|
||
* The edge workers' API (REBUILD.md, parts 1–2): a lease, the range's
|
||
* chunks as one stream, results per batch, and the end of the range.
|
||
*/
|
||
export default class NodeController {
|
||
private node(ctx: HttpContext) {
|
||
return Number(ctx.auth.use('node').getUserOrFail().id)
|
||
}
|
||
|
||
async lease(ctx: HttpContext) {
|
||
const lease = await Leases.lease(this.node(ctx))
|
||
if (!lease) return ctx.response.noContent()
|
||
return lease
|
||
}
|
||
|
||
/** The range's chunks from its resume point, waiting for those still being fetched. */
|
||
async stream(ctx: HttpContext) {
|
||
const nodeId = this.node(ctx)
|
||
const rangeId = Number(ctx.params.id)
|
||
const range = await this.leased(ctx, rangeId, nodeId)
|
||
if (!range) return
|
||
const spans = Spool.chunksOf({ start: Number(range.start_id), end: Number(range.end_id) }, Number(range.next_id))
|
||
async function* frames() {
|
||
for (const span of spans) {
|
||
const waitUntil = Date.now() + CHUNK_WAIT_MS
|
||
let quietSince = Date.now()
|
||
while (!(await Spool.has(rangeId, span))) {
|
||
if (Date.now() > waitUntil) throw new Error(`Chunk ${span.from}–${span.to} of range ${rangeId} wasn't spooled in time`)
|
||
if (Date.now() - quietSince >= KEEPALIVE_MS) {
|
||
quietSince = Date.now()
|
||
yield encodeFrame(KEEPALIVE)
|
||
await Leases.heartbeat(rangeId, nodeId)
|
||
}
|
||
await sleep(500)
|
||
}
|
||
yield encodeFrame({ ...span, data: await Spool.read(rangeId, span) })
|
||
// Streaming keeps the lease alive.
|
||
await Leases.heartbeat(rangeId, nodeId)
|
||
}
|
||
}
|
||
ctx.response.header('Content-Type', 'application/octet-stream')
|
||
ctx.response.stream(Readable.from(frames()))
|
||
}
|
||
|
||
async batch(ctx: HttpContext) {
|
||
const nodeId = this.node(ctx)
|
||
const rangeId = Number(ctx.params.id)
|
||
try {
|
||
return await BatchStore.store(rangeId, nodeId, fromWire(ctx.request.body() as WireBatch))
|
||
} catch (error) {
|
||
if (error instanceof StaleBatch) return ctx.response.conflict({ error: error.message })
|
||
throw error
|
||
}
|
||
}
|
||
|
||
async heartbeat(ctx: HttpContext) {
|
||
return this.lifecycle(ctx, (rangeId, nodeId) => Leases.heartbeat(rangeId, nodeId))
|
||
}
|
||
|
||
async finish(ctx: HttpContext) {
|
||
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))
|
||
}
|
||
|
||
private async lifecycle(ctx: HttpContext, action: (rangeId: number, nodeId: number) => Promise<void>) {
|
||
try {
|
||
await action(Number(ctx.params.id), this.node(ctx))
|
||
return { ok: true }
|
||
} catch (error) {
|
||
if (error instanceof LeaseLost) return ctx.response.conflict({ error: error.message })
|
||
throw error
|
||
}
|
||
}
|
||
|
||
private async leased(ctx: HttpContext, rangeId: number, nodeId: number) {
|
||
try {
|
||
return await Leases.leased(rangeId, nodeId)
|
||
} catch (error) {
|
||
if (error instanceof LeaseLost) {
|
||
ctx.response.conflict({ error: error.message })
|
||
return null
|
||
}
|
||
throw error
|
||
}
|
||
}
|
||
}
|