fix(code-runtime): bound completion measurement

This commit is contained in:
Tianyi Cui
2026-07-22 01:42:41 +08:00
parent 0708bd5ef2
commit bb6709b14a
7 changed files with 24 additions and 15 deletions
@@ -7,6 +7,7 @@
import { inspect } from 'node:util'
import type { DoneMessage, ReplyMessage, WorkerBootData, WorkerToHost } from './protocol.ts'
import { jsonValueBytesUpTo } from './output-json.ts'
import { snapshotCodeJsonValue } from './worker-json.ts'
/** The port surface the bootstrap needs — satisfied by `parentPort` and by the tests' fake. */
@@ -155,7 +156,7 @@ export function truncateUtf8Bytes(text: string, maxBytes: number): string {
*/
export function prepareCompletion(value: unknown, maxOutputBytes: number): Omit<DoneMessage, 'type'> {
if (value === undefined) return {}
let snapshot: unknown
let snapshot: ReturnType<typeof snapshotCodeJsonValue>
try {
snapshot = snapshotCodeJsonValue(value)
} catch {
@@ -164,8 +165,7 @@ export function prepareCompletion(value: unknown, maxOutputBytes: number): Omit<
if (snapshot === undefined) {
return { error: { kind: 'invalid-output', message: 'program completion must be lossless JSON' } }
}
const size = Buffer.byteLength(JSON.stringify(snapshot), 'utf8')
if (size > maxOutputBytes) {
if (jsonValueBytesUpTo(snapshot, maxOutputBytes) === undefined) {
return { error: { kind: 'output-limit', message: `outer output exceeded ${maxOutputBytes} bytes` } }
}
return { value: snapshot }
@@ -122,6 +122,7 @@ function waitForPipeDrain(stream: Readable): Promise<void> {
stream.once('error', done)
// Close the event-registration race if termination finished between the
// initial state check and the listeners above.
/* v8 ignore next -- this race cannot be scheduled deterministically between the adjacent state check and listener registration. */
if (stream.readableEnded || stream.destroyed) done()
})
}
@@ -362,12 +363,13 @@ export class WorkerCodeRuntime extends CodeRuntime {
// that were already queued; `finish` materializes the result only after
// termination completes.
const captureStray = (chunk: Buffer): void => {
/* v8 ignore next -- a second post-overflow chunk races immediate worker termination; the first overflow path is covered. */
if (terminalOverride !== undefined) return
const text = chunk.toString('utf8')
if (!output.admit(text, strayLogs)) {
const limited = output.limit([...logs, ...strayLogs, text])
terminalOverride = limited
finish(() => limited)
finish(limited)
}
}
worker.stdout.on('data', captureStray)
@@ -377,7 +379,7 @@ export class WorkerCodeRuntime extends CodeRuntime {
// logs captured before timeout, abort, or failure remain in the result.
let finishResolve!: () => void
const finished = new Promise<void>((done) => { finishResolve = done })
const finish = (finalize: () => CodeRunResult): void => {
const finish = (finalize: CodeRunResult | (() => CodeRunResult)): void => {
if (settled) return
settled = true
clearInterval(eluTimer)
@@ -390,7 +392,7 @@ export class WorkerCodeRuntime extends CodeRuntime {
const stdoutDrained = waitForPipeDrain(worker.stdout)
const stderrDrained = waitForPipeDrain(worker.stderr)
await Promise.all([worker.terminate(), stdoutDrained, stderrDrained])
const result = terminalOverride ?? finalize()
const result = terminalOverride ?? (typeof finalize === 'function' ? finalize() : finalize)
finishResolve()
resolve(result)
})
@@ -474,14 +476,12 @@ export class WorkerCodeRuntime extends CodeRuntime {
if (!message) return
if (message.type === 'log' && !settled && !output.admit(message.text, logs)) {
const limited = output.limit([...logs, ...strayLogs, message.text])
terminalOverride = limited
finish(() => limited)
finish(limited)
return
}
if (message.type === 'output-limit' && !settled) {
const limited = output.limit([...logs, ...strayLogs])
terminalOverride = limited
finish(() => limited)
finish(limited)
return
}
onCall(message)
@@ -219,6 +219,15 @@ describe('WorkerCodeRuntime — budgets and containment (real workers)', () => {
expect(after.value).toBe('alive')
}, 30_000)
it('reports a worker that exits before publishing a completion', async () => {
const { runtime } = await setup()
const result = await runtime.run({ program: 'process.exit(7)', bindings: [] })
expect(result).toEqual({
logs: [],
error: { kind: 'worker-exit', message: 'worker exited with code 7 before completing' },
})
})
it('fails runaway log output explicitly while retaining a bounded prefix', async () => {
const { runtime } = await setup({ maxOutputBytes: 300 })
const result = await runtime.run({
@@ -13,7 +13,7 @@ it('boots the source worker without workspace package outputs', async () => {
const directory = await mkdtemp(join(tmpdir(), 'dsh-code-source-worker-'))
let worker: Worker | undefined
try {
const files = ['worker.ts', 'bootstrap.ts', 'protocol.ts', 'worker-json.ts']
const files = ['worker.ts', 'bootstrap.ts', 'protocol.ts', 'worker-json.ts', 'output-json.ts']
await Promise.all(files.map(async (file) => {
await copyFile(new URL(`../src/${file}`, import.meta.url), join(directory, file))
}))