feat: add the worker-thread code runtime (dsh-code-runtime-worker)
The shipped backend of the code-execution seam, per the Code Mode RFC's
worker-thread section: one fresh Node worker per run, executing the
model's TypeScript after a host-side type-strip (wrapped in an
async-function shell so top-level return/await parse, sliced back out
position-preserved), bindings bridged over the message port under
hostile-peer rules (own-property name lookup, at-most-once replies,
post-settlement drops, null-prototype namespaces), logs streamed eagerly
with an in-band truncation marker, and two independent budgets — measured
event-loop busy time (computeMs) plus a never-pausing wall ceiling
(maxWallMs) — funneling into worker.terminate(). env: {} and execArgv: []
keep the isolate hermetic; disposal aborts in-flight runs and awaits
worker exits.
The worker entry loads unbuilt via Node's native type stripping
(src/worker.ts, erasable-only) and ships built as a sibling tsdown bundle
(lib/worker.js); tests/built-lib.e2e.ts pins the built load path under
plain node and joins the built-artifact smoke gate. Unit suites cover the
bootstrap in-process (fake port) and the runtime over real workers,
per-file 100%.
This commit is contained in:
Tianyi Cui
@@ -0,0 +1,214 @@
|
||||
import { describe, expect, it } from 'vitest'
|
||||
import { EventEmitter } from 'node:events'
|
||||
import { LogBuffer, makeConsoleShim, makeNamespaces, captureStreamWrites, prepareValue, runWorkerMain, wireReplies } from '@deepseek-ai/dsh-code-runtime-worker/src/bootstrap.ts'
|
||||
import type { BootstrapPort, PatchableStream, PendingCall } from '@deepseek-ai/dsh-code-runtime-worker/src/bootstrap.ts'
|
||||
import type { ReplyMessage, WorkerToHost } from '@deepseek-ai/dsh-code-runtime-worker/src/protocol.ts'
|
||||
import type { CodeLogEntry } from '@deepseek-ai/dsh-code-runtime'
|
||||
|
||||
/**
|
||||
* An in-process stand-in for the worker's parentPort: the test plays the
|
||||
* HOST side — inspect what the bootstrap posted, feed replies back — so
|
||||
* every line of worker-side logic runs under coverage without spawning an
|
||||
* isolate (real-worker behavior is pinned by runtime.spec.ts).
|
||||
*/
|
||||
class FakePort implements BootstrapPort {
|
||||
sent: WorkerToHost[] = []
|
||||
private readonly emitter = new EventEmitter()
|
||||
/** Host-scripted responder; return undefined to leave the call pending. */
|
||||
respond: (message: WorkerToHost) => ReplyMessage | undefined = () => undefined
|
||||
|
||||
postMessage(message: WorkerToHost): void {
|
||||
this.sent.push(message)
|
||||
const reply = this.respond(message)
|
||||
if (reply) queueMicrotask(() => this.emitter.emit('message', reply))
|
||||
}
|
||||
|
||||
on(event: 'message', listener: (message: ReplyMessage) => void): void {
|
||||
this.emitter.on(event, listener)
|
||||
}
|
||||
|
||||
deliver(message: ReplyMessage): void {
|
||||
this.emitter.emit('message', message)
|
||||
}
|
||||
|
||||
logs(): CodeLogEntry[] {
|
||||
return this.sent.filter(message => message.type === 'log').map(message => message.entry)
|
||||
}
|
||||
|
||||
done(): WorkerToHost | undefined {
|
||||
return this.sent.find(message => message.type === 'done')
|
||||
}
|
||||
}
|
||||
|
||||
function fakeStreams(): { stdout: PatchableStream; stderr: PatchableStream } {
|
||||
return { stdout: { write: () => true }, stderr: { write: () => true } }
|
||||
}
|
||||
|
||||
const BOOT = { maxLogBytes: 65_536, maxValueBytes: 32_768 }
|
||||
|
||||
describe('LogBuffer', () => {
|
||||
it('streams entries to the sink until the byte budget, then emits one marker and drops the rest', () => {
|
||||
const seen: CodeLogEntry[] = []
|
||||
const buffer = new LogBuffer(10, entry => seen.push(entry))
|
||||
buffer.push({ source: 'console', level: 'log', text: '12345' })
|
||||
buffer.push({ source: 'console', level: 'log', text: '123456' })
|
||||
buffer.push({ source: 'console', level: 'log', text: 'dropped' })
|
||||
expect(seen.map(entry => entry.text)).toEqual([
|
||||
'12345',
|
||||
'[dsh-code-runtime-worker] log capture truncated at 10 bytes',
|
||||
])
|
||||
})
|
||||
})
|
||||
|
||||
describe('makeConsoleShim', () => {
|
||||
it('captures the five levels and renders non-strings inspect-style', () => {
|
||||
const seen: CodeLogEntry[] = []
|
||||
const shim = makeConsoleShim(new LogBuffer(1_000, entry => seen.push(entry)))
|
||||
shim.log('plain', { a: 1 })
|
||||
shim.info('i')
|
||||
shim.warn('w')
|
||||
shim.error('e')
|
||||
shim.debug('d')
|
||||
expect(seen.map(entry => entry.level)).toEqual(['log', 'info', 'warn', 'error', 'debug'])
|
||||
expect(seen[0]?.text).toBe('plain { a: 1 }')
|
||||
expect(seen.every(entry => entry.source === 'console')).toBe(true)
|
||||
})
|
||||
})
|
||||
|
||||
describe('captureStreamWrites', () => {
|
||||
it('redirects writes into the buffer and restores on request', () => {
|
||||
const seen: CodeLogEntry[] = []
|
||||
const buffer = new LogBuffer(1_000, entry => seen.push(entry))
|
||||
let underlying = ''
|
||||
const stream: PatchableStream = { write: (chunk: unknown) => { underlying += String(chunk); return true } }
|
||||
const restore = captureStreamWrites(buffer, stream, 'stdout')
|
||||
stream.write('captured', 'utf8')
|
||||
stream.write(Buffer.from('bytes'))
|
||||
restore()
|
||||
stream.write('after')
|
||||
expect(seen.map(entry => entry.text)).toEqual(['captured', 'bytes'])
|
||||
expect(seen[0]).toMatchObject({ source: 'stdout' })
|
||||
expect(underlying).toBe('after')
|
||||
})
|
||||
})
|
||||
|
||||
describe('prepareValue', () => {
|
||||
it('omits undefined, passes small cloneable values raw', () => {
|
||||
expect(prepareValue(undefined, 100)).toEqual({})
|
||||
expect(prepareValue({ a: [1, 'two'] }, 100)).toEqual({ value: { a: [1, 'two'] } })
|
||||
})
|
||||
|
||||
it('replaces a non-cloneable value with its rendering', () => {
|
||||
const { value } = prepareValue({ fn: () => 1 }, 1_000)
|
||||
expect(typeof value).toBe('string')
|
||||
expect(value).toContain('fn')
|
||||
})
|
||||
|
||||
it('replaces an oversized value with a truncation-marked capped rendering', () => {
|
||||
const { value } = prepareValue('x'.repeat(50), 10)
|
||||
expect(value).toBe(`${'x'.repeat(10)}… [truncated]`)
|
||||
})
|
||||
})
|
||||
|
||||
describe('makeNamespaces', () => {
|
||||
it('exposes prototype-colliding names as ordinary own properties', async () => {
|
||||
const port = new FakePort()
|
||||
port.respond = message => message.type === 'call' ? { type: 'reply', id: message.id, ok: true, value: `${message.name}-ok` } : undefined
|
||||
const pending = new Map<number, PendingCall>()
|
||||
wireReplies(port, pending)
|
||||
const [tools] = makeNamespaces({ namespaces: [{ global: 'tools', names: ['__proto__', 'constructor', 'toString'] }] }, port, pending, { value: 1 }) as [Record<string, (args: unknown) => Promise<unknown>>]
|
||||
expect(Object.getPrototypeOf(tools)).toBeNull()
|
||||
await expect(tools['__proto__']?.({})).resolves.toBe('__proto__-ok')
|
||||
await expect(tools['constructor']?.({})).resolves.toBe('constructor-ok')
|
||||
await expect(tools['toString']?.({})).resolves.toBe('toString-ok')
|
||||
})
|
||||
|
||||
it('rejects a non-cloneable argument without leaking the pending entry', async () => {
|
||||
let firstCall = true
|
||||
const throwingPort: BootstrapPort = {
|
||||
// First call throws an Error (the real DataCloneError shape), the
|
||||
// second a bare string — the rejection renders both.
|
||||
postMessage: () => {
|
||||
if (firstCall) { firstCall = false; throw new Error('DataCloneError-ish') }
|
||||
throw 'raw-clone-failure'
|
||||
},
|
||||
on: () => {},
|
||||
}
|
||||
const pending = new Map<number, PendingCall>()
|
||||
const [tools] = makeNamespaces({ namespaces: [{ global: 'tools', names: ['x'] }] }, throwingPort, pending, { value: 1 }) as [Record<string, (args: unknown) => Promise<unknown>>]
|
||||
await expect(tools.x?.(() => 1)).rejects.toThrow(/structured-cloneable: DataCloneError-ish/)
|
||||
await expect(tools.x?.(() => 1)).rejects.toThrow(/structured-cloneable: raw-clone-failure/)
|
||||
expect(pending.size).toBe(0)
|
||||
})
|
||||
})
|
||||
|
||||
describe('runWorkerMain', () => {
|
||||
it('runs a program end-to-end: bindings, console, return value', async () => {
|
||||
const port = new FakePort()
|
||||
port.respond = message => message.type === 'call' ? { type: 'reply', id: message.id, ok: true, value: (message.args as { n: number }).n * 2 } : undefined
|
||||
await runWorkerMain(port, {
|
||||
...BOOT,
|
||||
code: 'const doubled = await tools.double({ n: 21 }); console.log("got", doubled); return { doubled };',
|
||||
namespaces: [{ global: 'tools', names: ['double'] }],
|
||||
}, fakeStreams())
|
||||
expect(port.logs()).toEqual([{ source: 'console', level: 'log', text: 'got 42' }])
|
||||
expect(port.done()).toEqual({ type: 'done', value: { doubled: 42 } })
|
||||
})
|
||||
|
||||
it('reports a thrown program error on the done message', async () => {
|
||||
const port = new FakePort()
|
||||
await runWorkerMain(port, { ...BOOT, code: 'throw new Error("boom")', namespaces: [] }, fakeStreams())
|
||||
const done = port.done()
|
||||
expect(done?.type).toBe('done')
|
||||
expect(done?.type === 'done' ? done.error?.message : undefined).toContain('boom')
|
||||
expect(done?.type === 'done' ? done.value : undefined).toBeUndefined()
|
||||
})
|
||||
|
||||
it('renders non-Error throws and stack-less Errors on the done message', async () => {
|
||||
const rawPort = new FakePort()
|
||||
await runWorkerMain(rawPort, { ...BOOT, code: 'throw "raw-throw"', namespaces: [] }, fakeStreams())
|
||||
expect(rawPort.done()).toEqual({ type: 'done', error: { message: 'raw-throw' } })
|
||||
|
||||
const barePort = new FakePort()
|
||||
await runWorkerMain(barePort, { ...BOOT, code: 'const e = new Error("bare"); e.stack = undefined; throw e', namespaces: [] }, fakeStreams())
|
||||
expect(barePort.done()).toEqual({ type: 'done', error: { message: 'bare' } })
|
||||
})
|
||||
|
||||
it('surfaces a host failure reply as a program-side rejection it can catch', async () => {
|
||||
const port = new FakePort()
|
||||
port.respond = message => message.type === 'call' ? { type: 'reply', id: message.id, ok: false, message: 'denied by host' } : undefined
|
||||
await runWorkerMain(port, {
|
||||
...BOOT,
|
||||
code: 'try { await tools.x({}) } catch (error) { return `caught: ${error.message}` }',
|
||||
namespaces: [{ global: 'tools', names: ['x'] }],
|
||||
}, fakeStreams())
|
||||
expect(port.done()).toEqual({ type: 'done', value: 'caught: denied by host' })
|
||||
})
|
||||
|
||||
it('ignores replies for unknown pending ids', async () => {
|
||||
const port = new FakePort()
|
||||
port.respond = (message) => {
|
||||
if (message.type !== 'call') return undefined
|
||||
// Deliver a stray reply first; the real one follows.
|
||||
port.deliver({ type: 'reply', id: 9_999, ok: true, value: 'stray' })
|
||||
return { type: 'reply', id: message.id, ok: true, value: 'real' }
|
||||
}
|
||||
await runWorkerMain(port, {
|
||||
...BOOT,
|
||||
code: 'return await tools.x({})',
|
||||
namespaces: [{ global: 'tools', names: ['x'] }],
|
||||
}, fakeStreams())
|
||||
expect(port.done()).toEqual({ type: 'done', value: 'real' })
|
||||
})
|
||||
|
||||
it('captures raw stream writes through the patched process streams', async () => {
|
||||
const port = new FakePort()
|
||||
const streams = fakeStreams()
|
||||
await runWorkerMain(port, { ...BOOT, code: 'return 1', namespaces: [] }, streams)
|
||||
streams.stdout.write('never seen — already restored? no: patch persists in worker')
|
||||
// The patch stays installed for the worker's lifetime; writes during the
|
||||
// program landed in order. Here the program wrote nothing via streams, so
|
||||
// only the post-run write above went through the patched slot.
|
||||
expect(port.logs().at(-1)).toMatchObject({ source: 'stdout' })
|
||||
})
|
||||
})
|
||||
@@ -0,0 +1,55 @@
|
||||
import { spawn } from 'node:child_process'
|
||||
import { existsSync } from 'node:fs'
|
||||
import { join } from 'node:path'
|
||||
import { fileURLToPath } from 'node:url'
|
||||
import { describe, expect, it } from 'vitest'
|
||||
|
||||
/**
|
||||
* BUILT-ARTIFACT smoke for the published package (the real-load-path guard
|
||||
* from docs/testing.md): the unit suite runs `src/` under vitest, where the
|
||||
* worker entry resolves to `src/worker.ts` — a consumer runs `lib/index.js`
|
||||
* under plain `node`, where it must resolve the sibling `lib/worker.js`
|
||||
* bundle instead. This spawns plain `node` (NOT tsx) from inside the package
|
||||
* directory and imports the package BY NAME, so resolution flows through the
|
||||
* real `exports` map exactly as it would from a downstream install; the
|
||||
* program exercises the type-strip, the worker spawn, the binding bridge,
|
||||
* and log capture end-to-end through the built bundles.
|
||||
*
|
||||
* It build-gates: SKIPS when the built artifacts are absent (suite run
|
||||
* without `pnpm run build`); CI runs it after the build step. KEYLESS — no
|
||||
* model is involved.
|
||||
*/
|
||||
|
||||
const pkgDir = fileURLToPath(new URL('..', import.meta.url))
|
||||
const built = ['lib/index.js', 'lib/worker.js'].every(file => existsSync(join(pkgDir, file)))
|
||||
&& existsSync(join(pkgDir, '../code-runtime/lib/index.js'))
|
||||
|
||||
describe.skipIf(!built)('built lib real load path (plain node)', () => {
|
||||
it('runs a TypeScript program with a binding through lib/index.js and its lib/worker.js entry', async () => {
|
||||
const script = `
|
||||
const { Context } = await import('cordis')
|
||||
const { WorkerCodeRuntime } = await import('@deepseek-ai/dsh-code-runtime-worker')
|
||||
const ctx = new Context()
|
||||
await ctx.plugin(WorkerCodeRuntime, {})
|
||||
const result = await ctx.codeRuntime.run({
|
||||
program: 'const doubled: number = await tools.double({ n: 21 }); console.log("halfway", doubled); return doubled;',
|
||||
bindings: [{ global: 'tools', functions: { double: async args => args.n * 2 } }],
|
||||
})
|
||||
console.log(JSON.stringify(result))
|
||||
process.exit(0)
|
||||
`
|
||||
const child = spawn(process.execPath, ['--input-type=module', '-e', script], { cwd: pkgDir, stdio: ['ignore', 'pipe', 'pipe'] })
|
||||
let stdout = ''
|
||||
let stderr = ''
|
||||
child.stdout.on('data', (chunk: Buffer) => { stdout += chunk.toString('utf8') })
|
||||
child.stderr.on('data', (chunk: Buffer) => { stderr += chunk.toString('utf8') })
|
||||
const exitCode = await new Promise<number | null>(resolve => child.on('close', resolve))
|
||||
|
||||
expect(exitCode, `stderr:\n${stderr}`).toBe(0)
|
||||
const lastLine = stdout.trim().split('\n').at(-1) ?? ''
|
||||
const result = JSON.parse(lastLine) as { value?: unknown; logs: { source: string; level?: string; text: string }[]; error?: unknown }
|
||||
expect(result.error).toBeUndefined()
|
||||
expect(result.value).toBe(42)
|
||||
expect(result.logs).toContainEqual({ source: 'console', level: 'log', text: 'halfway 42' })
|
||||
})
|
||||
})
|
||||
@@ -0,0 +1,328 @@
|
||||
import { describe, expect, it } from 'vitest'
|
||||
import { Context } from 'cordis'
|
||||
import { WorkerCodeRuntime } from '@deepseek-ai/dsh-code-runtime-worker'
|
||||
import type { Config } from '@deepseek-ai/dsh-code-runtime-worker'
|
||||
import type { CodeRunResult } from '@deepseek-ai/dsh-code-runtime'
|
||||
|
||||
/**
|
||||
* Integration suite over REAL worker threads (no mocks — workers are cheap
|
||||
* and local, per docs/testing.md's real-over-mock policy). Each test builds
|
||||
* a fresh context so budgets can be tuned per case.
|
||||
*/
|
||||
async function setup(config: Config = {}) {
|
||||
const ctx = new Context()
|
||||
await ctx.plugin(WorkerCodeRuntime, config)
|
||||
const runtime = ctx.codeRuntime as WorkerCodeRuntime
|
||||
return { ctx, runtime }
|
||||
}
|
||||
|
||||
/** Convenience: one namespace `tools` with the given functions. */
|
||||
function tools(functions: Record<string, (args: unknown) => Promise<unknown>>) {
|
||||
return [{ global: 'tools', functions }]
|
||||
}
|
||||
|
||||
describe('WorkerCodeRuntime — programs and bindings (real workers)', () => {
|
||||
it('registers with the seam descriptors', async () => {
|
||||
const { runtime } = await setup()
|
||||
expect(runtime.language).toBe('typescript')
|
||||
expect(runtime.isolation).toBe('worker-thread')
|
||||
})
|
||||
|
||||
it('runs TypeScript (erasable syntax), captures console/stdout in order, returns the value', async () => {
|
||||
const { runtime } = await setup()
|
||||
const result = await runtime.run({
|
||||
program: `
|
||||
interface Point { x: number; y: number }
|
||||
const p: Point = { x: 1, y: 2 } as Point;
|
||||
console.log('point', p);
|
||||
process.stdout.write('raw-out\\n');
|
||||
console.warn('careful');
|
||||
return p.x + p.y;
|
||||
`,
|
||||
bindings: [],
|
||||
})
|
||||
expect(result.error).toBeUndefined()
|
||||
expect(result.value).toBe(3)
|
||||
expect(result.logs.map(entry => [entry.source, entry.level ?? null])).toEqual([
|
||||
['console', 'log'],
|
||||
['stdout', null],
|
||||
['console', 'warn'],
|
||||
])
|
||||
expect(result.logs[0]?.text).toBe('point { x: 1, y: 2 }')
|
||||
})
|
||||
|
||||
it('bridges binding calls both ways and rejects the program-side call on a host rejection', async () => {
|
||||
const { runtime } = await setup()
|
||||
const calls: unknown[] = []
|
||||
const result = await runtime.run({
|
||||
program: `
|
||||
const first = await tools.echo({ n: 1 });
|
||||
let caught = '';
|
||||
try { await tools.fail({}) } catch (error) { caught = error.message }
|
||||
let caughtRaw = '';
|
||||
try { await tools.failRaw({}) } catch (error) { caughtRaw = error.message }
|
||||
return { first, caught, caughtRaw };
|
||||
`,
|
||||
bindings: tools({
|
||||
echo: async (args) => { calls.push(args); return { echoed: args } },
|
||||
fail: async () => { throw new Error('nope') },
|
||||
// A non-Error throw: the host renders it, the program still catches.
|
||||
failRaw: async () => { throw 'raw-nope' },
|
||||
}),
|
||||
})
|
||||
expect(result.error).toBeUndefined()
|
||||
expect(result.value).toEqual({ first: { echoed: { n: 1 } }, caught: 'nope', caughtRaw: 'raw-nope' })
|
||||
expect(calls).toEqual([{ n: 1 }])
|
||||
})
|
||||
|
||||
it('reports non-erasable syntax as an exception without spawning a worker', async () => {
|
||||
const { runtime } = await setup()
|
||||
const result = await runtime.run({ program: 'enum E { A }\nreturn 1', bindings: [] })
|
||||
expect(result.error?.kind).toBe('exception')
|
||||
expect(result.error?.message).toMatch(/enum|strip/i)
|
||||
})
|
||||
|
||||
it('reports a runtime throw as an exception with the message', async () => {
|
||||
const { runtime } = await setup()
|
||||
const result = await runtime.run({ program: 'throw new Error("kaboom")', bindings: [] })
|
||||
expect(result.error?.kind).toBe('exception')
|
||||
expect(result.error?.message).toContain('kaboom')
|
||||
})
|
||||
|
||||
it('gives the program an EMPTY environment', async () => {
|
||||
const { runtime } = await setup()
|
||||
const result = await runtime.run({ program: 'return JSON.stringify(process.env)', bindings: [] })
|
||||
expect(result.value).toBe('{}')
|
||||
})
|
||||
|
||||
it('replaces a non-cloneable return value with a string rendering', async () => {
|
||||
const { runtime } = await setup()
|
||||
const result = await runtime.run({ program: 'return { f: () => 1 }', bindings: [] })
|
||||
expect(typeof result.value).toBe('string')
|
||||
})
|
||||
|
||||
it('keeps logs streamed before a failure', async () => {
|
||||
const { runtime } = await setup()
|
||||
const result = await runtime.run({
|
||||
program: 'console.log("before"); throw new Error("after-log")',
|
||||
bindings: [],
|
||||
})
|
||||
expect(result.error?.kind).toBe('exception')
|
||||
expect(result.logs.map(entry => entry.text)).toContain('before')
|
||||
})
|
||||
})
|
||||
|
||||
describe('WorkerCodeRuntime — budgets and containment (real workers)', () => {
|
||||
it('ends a hot loop at the compute budget — including behind a pending decoy dispatch', async () => {
|
||||
const { runtime } = await setup({ computeMs: 300, maxWallMs: 30_000 })
|
||||
const result = await runtime.run({
|
||||
// The decoy: fire a call at a never-resolving binding WITHOUT awaiting,
|
||||
// then spin. Host-side pending-call bookkeeping would pause a naive
|
||||
// budget here; measured busy time cannot be fooled.
|
||||
program: 'void tools.slow({}); for (;;) {}',
|
||||
bindings: tools({ slow: () => new Promise(() => {}) }),
|
||||
})
|
||||
expect(result.error?.kind).toBe('timeout')
|
||||
expect(result.error?.message).toContain('compute budget')
|
||||
}, 15_000)
|
||||
|
||||
it('does not charge time spent awaiting a slow binding against the compute budget', async () => {
|
||||
const { runtime } = await setup({ computeMs: 250, maxWallMs: 30_000 })
|
||||
const result = await runtime.run({
|
||||
program: 'return await tools.slow({})',
|
||||
bindings: tools({ slow: () => new Promise(resolve => setTimeout(() => { resolve('slow-done') }, 700)) }),
|
||||
})
|
||||
expect(result.error).toBeUndefined()
|
||||
expect(result.value).toBe('slow-done')
|
||||
}, 15_000)
|
||||
|
||||
it('ends an idle-forever run at the wall-clock ceiling', async () => {
|
||||
const { runtime } = await setup({ computeMs: 30_000, maxWallMs: 400 })
|
||||
const result = await runtime.run({
|
||||
program: 'await tools.never({}); return 1',
|
||||
bindings: tools({ never: () => new Promise(() => {}) }),
|
||||
})
|
||||
expect(result.error?.kind).toBe('timeout')
|
||||
expect(result.error?.message).toContain('wall-clock ceiling')
|
||||
}, 15_000)
|
||||
|
||||
it('reports an abort mid-run and stops the worker', async () => {
|
||||
const { runtime } = await setup()
|
||||
const controller = new AbortController()
|
||||
setTimeout(() => { controller.abort('user-cancel') }, 150)
|
||||
const result = await runtime.run({ program: 'for (;;) {}', bindings: [], signal: controller.signal })
|
||||
expect(result.error).toEqual({ kind: 'abort', message: 'user-cancel' })
|
||||
}, 15_000)
|
||||
|
||||
it('reports a pre-aborted signal without spawning', async () => {
|
||||
const { runtime } = await setup()
|
||||
const controller = new AbortController()
|
||||
controller.abort('too-late')
|
||||
const result = await runtime.run({ program: 'return 1', bindings: [], signal: controller.signal })
|
||||
expect(result.error).toEqual({ kind: 'abort', message: 'too-late' })
|
||||
})
|
||||
|
||||
it('drops a binding resolution that lands after the run settled', async () => {
|
||||
const { runtime } = await setup()
|
||||
const controller = new AbortController()
|
||||
let replyDelivered!: Promise<void>
|
||||
const result = await runtime.run({
|
||||
program: 'void tools.late({}); for (;;) {}',
|
||||
bindings: tools({
|
||||
// Anchored on invocation: abort 100ms after the call reaches the
|
||||
// host, resolve 400ms after — by then the run has settled, so the
|
||||
// resolution's reply hits the post-settlement drop.
|
||||
late: () => new Promise((resolve) => {
|
||||
setTimeout(() => { controller.abort('cancel-now') }, 100)
|
||||
replyDelivered = new Promise(done => setTimeout(() => { resolve('too-late'); done() }, 400))
|
||||
}),
|
||||
}),
|
||||
signal: controller.signal,
|
||||
})
|
||||
expect(result.error).toEqual({ kind: 'abort', message: 'cancel-now' })
|
||||
// Let the late resolution actually fire so its reply executes instead of
|
||||
// being cancelled with the test.
|
||||
await replyDelivered
|
||||
}, 15_000)
|
||||
|
||||
it('contains an OOM under resourceLimits as worker-exit, host process healthy', async () => {
|
||||
const { runtime } = await setup({ maxOldGenerationSizeMb: 32 })
|
||||
const result = await runtime.run({
|
||||
program: 'const hog = []; for (;;) hog.push(new Array(1e6).fill(1));',
|
||||
bindings: [],
|
||||
})
|
||||
expect(result.error?.kind).toBe('worker-exit')
|
||||
// And the host is fine: run something else.
|
||||
const after = await runtime.run({ program: 'return "alive"', bindings: [] })
|
||||
expect(after.value).toBe('alive')
|
||||
}, 30_000)
|
||||
|
||||
it('truncates runaway log output at the byte budget with an in-band marker', async () => {
|
||||
const { runtime } = await setup({ maxLogBytes: 300 })
|
||||
const result = await runtime.run({
|
||||
program: 'for (let i = 0; i < 1000; i++) console.log("spam line", i); return 1',
|
||||
bindings: [],
|
||||
})
|
||||
expect(result.logs.at(-1)?.text).toContain('truncated at 300 bytes')
|
||||
const total = result.logs.reduce((sum, entry) => sum + Buffer.byteLength(entry.text, 'utf8'), 0)
|
||||
expect(total).toBeLessThan(1_000)
|
||||
})
|
||||
|
||||
it('caps an oversized return value with a truncation marker', async () => {
|
||||
const { runtime } = await setup({ maxValueBytes: 64 })
|
||||
const result = await runtime.run({ program: 'return "y".repeat(10_000)', bindings: [] })
|
||||
expect(result.value).toBe(`${'y'.repeat(64)}… [truncated]`)
|
||||
})
|
||||
|
||||
it('captures pipe writes that bypass the patched write slot as stray logs, capped by the same budget', async () => {
|
||||
const { runtime } = await setup({ maxLogBytes: 4 })
|
||||
const result = await runtime.run({
|
||||
// The bootstrap patches the stream instance's own `write`; going
|
||||
// through the prototype's slot reaches the real pipe underneath, so
|
||||
// the bytes arrive host-side as stray data. The pauses keep the two
|
||||
// writes in separate pipe chunks and let them land before settlement.
|
||||
program: `
|
||||
const write = (text) => Object.getPrototypeOf(process.stdout).write.call(process.stdout, text);
|
||||
write('abcd');
|
||||
await new Promise(resolve => setTimeout(resolve, 150));
|
||||
write('ef');
|
||||
await new Promise(resolve => setTimeout(resolve, 100));
|
||||
return 1;
|
||||
`,
|
||||
bindings: [],
|
||||
})
|
||||
expect(result.error).toBeUndefined()
|
||||
expect(result.logs).toContainEqual({ source: 'stdout', text: 'abcd' })
|
||||
expect(result.logs.map(entry => entry.text)).not.toContain('ef')
|
||||
}, 15_000)
|
||||
})
|
||||
|
||||
describe('WorkerCodeRuntime — hostile programs (real workers)', () => {
|
||||
it('survives forged port traffic: unknown binding names, duplicate ids, junk shapes', async () => {
|
||||
const { runtime } = await setup()
|
||||
const result = await runtime.run({
|
||||
program: `
|
||||
const { parentPort } = await import('node:worker_threads');
|
||||
parentPort.postMessage({ type: 'call', id: 7777, global: 'tools', name: 'missing', args: {} });
|
||||
parentPort.postMessage({ type: 'call', id: 7777, global: 'tools', name: 'missing', args: {} });
|
||||
parentPort.postMessage({ type: 'call', id: 7778, global: 'tools', name: 'constructor', args: {} });
|
||||
parentPort.postMessage({ type: 'junk' });
|
||||
return await tools.real({});
|
||||
`,
|
||||
bindings: tools({ real: async () => 'still-works' }),
|
||||
})
|
||||
expect(result.error).toBeUndefined()
|
||||
expect(result.value).toBe('still-works')
|
||||
})
|
||||
|
||||
it('answers a binding whose resolution cannot be cloned with a failure reply', async () => {
|
||||
const { runtime } = await setup()
|
||||
const result = await runtime.run({
|
||||
program: 'try { await tools.bad({}) } catch (error) { return error.message }',
|
||||
bindings: tools({ bad: async () => (() => 1) }),
|
||||
})
|
||||
expect(result.value).toContain('not structured-cloneable')
|
||||
})
|
||||
|
||||
it('exposes binding names that collide with Object.prototype as ordinary functions', async () => {
|
||||
const { runtime } = await setup()
|
||||
const result = await runtime.run({
|
||||
program: 'return [await tools["__proto__"]({}), await tools["constructor"]({}), typeof tools["hasOwnProperty"]]',
|
||||
// Computed keys: a literal `'__proto__': …` entry would SET the record's
|
||||
// prototype instead of declaring a binding of that name.
|
||||
bindings: tools({ ['__proto__']: async () => 'proto-ok', ['constructor']: async () => 'ctor-ok' }),
|
||||
})
|
||||
expect(result.value).toEqual(['proto-ok', 'ctor-ok', 'undefined'])
|
||||
})
|
||||
})
|
||||
|
||||
describe('WorkerCodeRuntime — seam misuse and lifecycle', () => {
|
||||
it('rejects invalid binding globals loudly (identifier, reserved word, duplicate, console)', async () => {
|
||||
const { runtime } = await setup()
|
||||
const cases: [string, RegExp][] = [
|
||||
['not valid!', /not a usable identifier/],
|
||||
['await', /not a usable identifier/],
|
||||
['console', /duplicate binding global/],
|
||||
]
|
||||
for (const [global, message] of cases) {
|
||||
await expect(runtime.run({ program: 'return 1', bindings: [{ global, functions: {} }] })).rejects.toThrow(message)
|
||||
}
|
||||
await expect(runtime.run({
|
||||
program: 'return 1',
|
||||
bindings: [{ global: 'tools', functions: {} }, { global: 'tools', functions: {} }],
|
||||
})).rejects.toThrow(/duplicate binding global/)
|
||||
})
|
||||
|
||||
it('rejects config values that are not positive numbers', async () => {
|
||||
const ctx = new Context()
|
||||
await expect(ctx.plugin(WorkerCodeRuntime, { computeMs: -1 })).rejects.toThrow(/positive number/)
|
||||
})
|
||||
|
||||
it('keeps runs isolated: no state survives from one run to the next', async () => {
|
||||
const { runtime } = await setup()
|
||||
await runtime.run({ program: 'globalThis.leak = "value"; return 1', bindings: [] })
|
||||
const second = await runtime.run({ program: 'return typeof globalThis.leak', bindings: [] })
|
||||
expect(second.value).toBe('undefined')
|
||||
})
|
||||
|
||||
it('disposal aborts in-flight runs, awaits worker exit, and rejects later runs', async () => {
|
||||
const ctx = new Context()
|
||||
const fiber = await ctx.plugin(WorkerCodeRuntime)
|
||||
const runtime = ctx.codeRuntime as WorkerCodeRuntime
|
||||
const inflight: Promise<CodeRunResult> = runtime.run({ program: 'for (;;) {}', bindings: [] })
|
||||
// Give the worker a moment to actually start spinning.
|
||||
await new Promise(resolve => setTimeout(resolve, 200))
|
||||
await fiber.dispose()
|
||||
const result = await inflight
|
||||
expect(result.error).toEqual({ kind: 'abort', message: 'runtime disposed' })
|
||||
await expect(runtime.run({ program: 'return 1', bindings: [] })).rejects.toThrow(/after disposal/)
|
||||
}, 15_000)
|
||||
|
||||
it('removes ctx.codeRuntime when the providing fiber disposes (HMR safety)', async () => {
|
||||
const ctx = new Context()
|
||||
const fiber = await ctx.plugin(WorkerCodeRuntime)
|
||||
expect(ctx.get('codeRuntime')).toBeInstanceOf(WorkerCodeRuntime)
|
||||
await fiber.dispose()
|
||||
expect(ctx.get('codeRuntime')).toBeUndefined()
|
||||
})
|
||||
})
|
||||
Reference in New Issue
Block a user