Daan MeijerandClaude Opus 5.5
fa7d4f3394
End to end (workstation, recent a.b.boneless ranges from the spool): a 2M range in 10–12 s, 166k–202k headers/s per worker process (edge:bench on one core: 215k). - Prefetcher: spooled_at was set against a total counted from the range's resume point, which shrinks as the worker advances: the last chunks were never fetched and the worker waited. Now counted against the whole range, only new chunks counted, and leased ranges are always checked against the files. Planned ranges ahead are bounded (it had spooled ever more); the spool size is measured every 30 s (a stat per file). - The stream sends an empty frame every 20 s while waiting for a chunk (Node's fetch gave up on a quiet body after 5 minutes). - A range taken over midway: the first batch starts at the resume point (it started at the range's start and was refused). - edge:node inflates the next chunk on the thread pool while indexing the current one, and doesn't wait for a post before indexing on (posts stay in order; a failed post stops the range). Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
103 lines
3.7 KiB
TypeScript
103 lines
3.7 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))
|
||
}
|
||
|
||
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
|
||
}
|
||
}
|
||
}
|