From f934347041c474a91cb57798ed1ec55b74502a94 Mon Sep 17 00:00:00 2001 From: "Timothy J. Aveni" Date: Wed, 7 Oct 2026 14:44:59 -0700 Subject: [PATCH 1/4] Bound checked candidate metadata and restricted invocation resources --- restricted-execution/service.mjs | 182 ++++++++++++++++++++------ restricted-execution/service.test.mjs | 105 ++++++++++++++- 2 files changed, 245 insertions(+), 42 deletions(-) diff --git a/restricted-execution/service.mjs b/restricted-execution/service.mjs index 4e5967d..b1ea480 100644 --- a/restricted-execution/service.mjs +++ b/restricted-execution/service.mjs @@ -1,6 +1,7 @@ import { spawn } from "node:child_process"; import { createServer, request as httpRequest } from "node:http"; -import { mkdtemp, readFile, writeFile, rm } from "node:fs/promises"; +import { mkdtemp, readFile, writeFile, rm, open } from "node:fs/promises"; +import { constants } from "node:fs"; import { tmpdir } from "node:os"; import { join, dirname } from "node:path"; @@ -12,6 +13,24 @@ const absolute = (value) => value.startsWith("/") && !/[\s\u0000-\u001f]/.test(value) && !value.split("/").includes(".."); +async function boundedSource(path, maximum, code) { + const handle = await open(path, constants.O_RDONLY | constants.O_NOFOLLOW | constants.O_NONBLOCK); + try { + const stat = await handle.stat(); + if (!stat.isFile() || stat.size > maximum) throw fixedError(code); + const bytes = Buffer.alloc(stat.size + 1); + let offset = 0; + while (offset < bytes.length) { + const read = await handle.read(bytes, offset, bytes.length - offset, offset); + if (!read.bytesRead) break; + offset += read.bytesRead; + } + if (offset !== stat.size) throw fixedError(code); + return new TextDecoder("utf-8", { fatal: true }).decode(bytes.subarray(0, offset)); + } finally { + await handle.close(); + } +} /** Trusted stdio adapter inside one workspace VM. Engine processes never own * coordinator credentials. The supervising service must impose VM/OS limits. @@ -47,6 +66,8 @@ export async function serveRestrictedRegistry(config, io = process) { maxPortCalls: config.maxPortCalls ?? 1000, deadlineMs: config.deadlineMs ?? 4000, maxQueued: config.maxQueued ?? 8, + maxActive: config.maxActive ?? 2, + maxNested: config.maxNested ?? 4, }; if ( !Number.isSafeInteger(limits.maxBytes) || @@ -60,17 +81,22 @@ export async function serveRestrictedRegistry(config, io = process) { limits.deadlineMs > 60000 || !Number.isSafeInteger(limits.maxQueued) || limits.maxQueued < 0 || - limits.maxQueued > 64 + limits.maxQueued > 64 || + !Number.isSafeInteger(limits.maxActive) || + limits.maxActive < 1 || + limits.maxActive > 16 || + !Number.isSafeInteger(limits.maxNested) || + limits.maxNested < 1 || + limits.maxNested > 16 ) throw fixedError("INVALID_RESTRICTED_LIMITS"); - const packageSource = await readFile(config.module, "utf8"); - if (Buffer.byteLength(packageSource) > 1024 * 1024) throw fixedError("RESTRICTED_SOURCE_LIMIT"); - const workerSource = await readFile(config.worker, "utf8"); - if (Buffer.byteLength(workerSource) > 65536) throw fixedError("RESTRICTED_ENTRY_LIMIT"); - const sdk = await Promise.all(config.sdk.map(async (m) => ({ ...m, source: await readFile(m.path, "utf8") }))); - if (sdk.some((m) => Buffer.byteLength(m.source) > 2 * 1024 * 1024)) throw fixedError("RESTRICTED_SDK_LIMIT"); - let active, - buffered = Buffer.alloc(0), + const packageSource = await boundedSource(config.module, 1024 * 1024, "RESTRICTED_SOURCE_LIMIT"); + const workerSource = await boundedSource(config.worker, 65536, "RESTRICTED_ENTRY_LIMIT"); + const sdk = []; + for (const m of config.sdk) + sdk.push({ ...m, source: await boundedSource(m.path, 2 * 1024 * 1024, "RESTRICTED_SDK_LIMIT") }); + const active = new Map(); + let buffered = Buffer.alloc(0), stopped = false; const queued = []; const send = (value) => { @@ -83,11 +109,40 @@ export async function serveRestrictedRegistry(config, io = process) { throw fixedError("RESTRICTED_BACKPRESSURE"); io.stdout.write(line); }; + // Linux USER_HZ is 100 on the supported NixOS architectures. Read process + // CPU outside the engine; authored code cannot forge these usage messages. + const sampleCPU = async (call) => { + if (!call.child?.pid) return; + let text; + try { + text = await readFile(`/proc/${call.child.pid}/stat`, "utf8"); + } catch (error) { + if (error.code === "ENOENT" || error.code === "ESRCH") return; + throw error; + } + const fields = text.slice(text.lastIndexOf(")") + 2).split(" "); + const cpuMicroseconds = (Number(fields[11]) + Number(fields[12])) * 10000; + if (!Number.isSafeInteger(cpuMicroseconds) || cpuMicroseconds < (call.cpuMicroseconds ?? 0)) + throw fixedError("RESTRICTED_CPU_ACCOUNTING"); + if (cpuMicroseconds !== (call.cpuMicroseconds ?? 0)) { + call.cpuMicroseconds = cpuMicroseconds; + send({ kind: "usage", invocation: call.message.invocation, frame: call.message.frame, cpuMicroseconds }); + } + }; + const watchCPU = (call) => { + call.cpuTimer = setTimeout(() => { + call.cpuRead = sampleCPU(call).then(() => { + if (call.open) watchCPU(call); + }); + void call.cpuRead.catch(fail); + }, 20); + }; const retire = async (call, reason) => { if (call.retiring) return call.retiring; call.open = false; call.retiring = (async () => { clearTimeout(call.timer); + clearTimeout(call.cpuTimer); const index = queued.indexOf(call); if (index >= 0) queued.splice(index, 1); for (const pending of call.pending.values()) pending.reject(reason); @@ -95,6 +150,14 @@ export async function serveRestrictedRegistry(config, io = process) { // Cancellation can arrive during async directory/socket preparation. // Wait for it to settle before claiming every process/handle is gone. await call.preparing?.catch(() => {}); + clearTimeout(call.cpuTimer); + await call.cpuRead?.catch(() => {}); + let usageFailure; + try { + await sampleCPU(call); + } catch (error) { + usageFailure = error; + } if (call.child?.pid) { try { process.kill(-call.child.pid, "SIGKILL"); @@ -106,7 +169,8 @@ export async function serveRestrictedRegistry(config, io = process) { call.server?.closeAllConnections(); if (call.server) await new Promise((resolve) => call.server.close(resolve)); if (call.directory) await rm(call.directory, { recursive: true, force: true }); - if (active === call) active = undefined; + active.delete(call.message.invocation); + if (usageFailure) throw usageFailure; })(); return call.retiring; }; @@ -115,7 +179,7 @@ export async function serveRestrictedRegistry(config, io = process) { stopped = true; io.stdin.destroy(); for (const call of [...queued]) await retire(call, error); - if (active) await retire(active, error); + await Promise.all([...active.values()].map((call) => retire(call, error))); io.stdout.end(); io.exitCode = 1; }; @@ -129,17 +193,27 @@ export async function serveRestrictedRegistry(config, io = process) { } return JSON.parse(new TextDecoder("utf-8", { fatal: true }).decode(Buffer.concat(chunks))); }; + const available = (call) => + [...active.values()].filter((entry) => entry.nested === call.nested).length < + (call.nested ? limits.maxNested : limits.maxActive); const pump = () => { - if (!stopped && !active && queued.length) void invoke(queued.shift()).catch(fail); + while (!stopped) { + const index = queued.findIndex(available); + if (index < 0) return; + void invoke(queued.splice(index, 1)[0]).catch(fail); + } }; const finish = async (call, reason, reply) => { - await retire(call, reason); - send(reply); - pump(); + if (reply.kind === "cancelled") call.cancellationReply = reply; + return (call.finished ??= (async () => { + await retire(call, reason); + send(call.cancellationReply ?? reply); + pump(); + })()); }; const invoke = async (call) => { const { message } = call; - active = call; + active.set(message.invocation, call); const terminate = (code) => { if (!call.open) return; void finish(call, fixedError(code), { @@ -169,7 +243,10 @@ export async function serveRestrictedRegistry(config, io = process) { !Object.hasOwn(port, "input") ) throw fixedError("INVALID_RESTRICTED_PORT"); - if (++call.sequence > limits.maxPortCalls || call.pending.size >= limits.maxPortCalls) + if ( + ++call.sequence > Math.min(limits.maxPortCalls, message.budget?.calls ?? limits.maxPortCalls) || + call.pending.size >= limits.maxPortCalls + ) throw fixedError("RESTRICTED_PORT_LIMIT"); const sequence = call.sequence; const result = new Promise((resolve, reject) => call.pending.set(sequence, { resolve, reject })); @@ -229,6 +306,7 @@ export async function serveRestrictedRegistry(config, io = process) { call.child.once("exit", resolve); call.child.once("error", resolve); }); + watchCPU(call); let diagnostics = 0; call.child.stderr.on("data", (chunk) => { if ((diagnostics += chunk.length) > 65536) terminate("RESTRICTED_DIAGNOSTIC_LIMIT"); @@ -246,15 +324,18 @@ export async function serveRestrictedRegistry(config, io = process) { reject(fixedError("EXPIRED_CONTEXT")); return; } - const request = httpRequest({ socketPath: engine, path: "/", method: "POST" }, async (response) => { - try { - const result = await body(response); - if (response.statusCode !== 200) throw fixedError("RESTRICTED_PACKAGE_FAILED"); - resolve(result); - } catch (error) { - reject(error); - } - }); + const request = httpRequest( + { socketPath: engine, path: "/", method: "POST", agent: false }, + async (response) => { + try { + const result = await body(response); + if (response.statusCode !== 200) throw fixedError("RESTRICTED_PACKAGE_FAILED"); + resolve(result); + } catch (error) { + reject(error); + } + }, + ); request.once("error", (error) => { if (call.open && ["ENOENT", "ECONNREFUSED"].includes(error.code)) setTimeout(attempt, 5); else reject(error); @@ -311,9 +392,30 @@ export async function serveRestrictedRegistry(config, io = process) { message.frame.length > 4096 ) throw fixedError("INVALID_RESTRICTED_INVOCATION"); - if ([active, ...queued].some((call) => call?.message.invocation === message.invocation)) + if ( + message.budget !== undefined && + (!object(message.budget) || + Object.keys(message.budget).sort().join(",") !== "calls,deadlineMs,depth,root" || + typeof message.budget.root !== "string" || + !message.budget.root || + message.budget.root.length > 4096 || + !Number.isSafeInteger(message.budget.depth) || + message.budget.depth < 0 || + message.budget.depth > 32 || + !Number.isSafeInteger(message.budget.calls) || + message.budget.calls < 1 || + message.budget.calls > 1000000 || + !Number.isSafeInteger(message.budget.deadlineMs) || + message.budget.deadlineMs < 1 || + message.budget.deadlineMs > 60000) + ) + throw fixedError("INVALID_RESTRICTED_BUDGET"); + if (active.has(message.invocation) || queued.some((call) => call.message.invocation === message.invocation)) throw fixedError("RESTRICTED_DUPLICATE_INVOCATION"); - if (active && queued.length >= limits.maxQueued) { + const call = { message, open: true, pending: new Map(), sequence: 0, nested: (message.budget?.depth ?? 0) > 1 }; + // Descendants cannot wait behind ancestors that hold all nested capacity. + // Reject excess fan-out/depth rather than deadlock an admitted call tree. + if (!available(call) && (call.nested || queued.length >= limits.maxQueued)) { send({ invocation: message.invocation, status: "failed", @@ -321,16 +423,18 @@ export async function serveRestrictedRegistry(config, io = process) { }); return; } - const call = { message, open: true, pending: new Map(), sequence: 0 }; // Deadline includes queue wait: overload cannot extend a transaction frame. - call.timer = setTimeout(() => { - if (!call.open) return; - void finish(call, fixedError("RESTRICTED_DEADLINE"), { - invocation: message.invocation, - status: "failed", - error: { code: "RESTRICTED_DEADLINE", message: "RESTRICTED_DEADLINE" }, - }).catch(fail); - }, limits.deadlineMs); + call.timer = setTimeout( + () => { + if (!call.open) return; + void finish(call, fixedError("RESTRICTED_DEADLINE"), { + invocation: message.invocation, + status: "failed", + error: { code: "RESTRICTED_DEADLINE", message: "RESTRICTED_DEADLINE" }, + }).catch(fail); + }, + Math.min(limits.deadlineMs, message.budget?.deadlineMs ?? limits.deadlineMs), + ); queued.push(call); pump(); return; @@ -346,7 +450,7 @@ export async function serveRestrictedRegistry(config, io = process) { }).catch(fail); return; } - const call = active; + const call = active.get(message.invocation); if (!call || message.invocation !== call.message.invocation || message.frame !== call.message.frame) { if (message.kind === "cancel") send({ kind: "cancelled", invocation: message.invocation, frame: message.frame }); else if (message.kind !== "port-result") throw fixedError("RESTRICTED_UNKNOWN_INVOCATION"); diff --git a/restricted-execution/service.test.mjs b/restricted-execution/service.test.mjs index b3634b2..e127ae8 100644 --- a/restricted-execution/service.test.mjs +++ b/restricted-execution/service.test.mjs @@ -19,6 +19,7 @@ async function fixture(source, options = {}) { const io = { stdin: new PassThrough(), stdout: new PassThrough(), exitCode: 0 }; const waiting = [], queued = []; + const usage = []; let buffer = ""; io.stdout.on("data", (chunk) => { buffer += chunk; @@ -26,6 +27,10 @@ async function fixture(source, options = {}) { while ((index = buffer.indexOf("\n")) >= 0) { const value = JSON.parse(buffer.slice(0, index)); buffer = buffer.slice(index + 1); + if (value.kind === "usage") { + usage.push(value); + continue; + } if (waiting.length) waiting.shift()(value); else queued.push(value); } @@ -50,7 +55,7 @@ async function fixture(source, options = {}) { ); assert.deepEqual(await receive(), { abi, kind: "ready", exports: ["test"] }); const send = (value) => io.stdin.write(JSON.stringify({ abi, ...value }) + "\n"); - const start = (input = null) => { + const start = (input = null, budget) => { const invocation = "invoke-" + ++next, frame = "frame-" + next; send({ @@ -62,6 +67,7 @@ async function fixture(source, options = {}) { receiver: null, input, effect: "snapshot", + ...(budget ? { budget } : {}), }); return { invocation, frame }; }; @@ -70,6 +76,7 @@ async function fixture(source, options = {}) { receive, send, start, + usage, async close() { await service.close(); await rm(directory, { recursive: true, force: true }); @@ -78,6 +85,94 @@ async function fixture(source, options = {}) { } const registry = (code) => `export const registry={exports:['test'],async invoke(request,channel){${code}}};`; +test("parallel frames route independently and cancellation cannot retire a neighbor", async () => { + const f = await fixture( + registry("return await channel({portId:'read',action:{kind:'state',primitive:'read'},input:request.input});"), + ); + try { + const first = f.start("a"), + second = f.start("b"); + const ports = new Map([await f.receive(), await f.receive()].map((port) => [port.invocation, port])); + assert.equal(ports.get(first.invocation).input, "a"); + assert.equal(ports.get(second.invocation).input, "b"); + f.send({ kind: "cancel", ...first }); + assert.equal((await f.receive()).kind, "cancelled"); + f.send({ + kind: "port-result", + ...first, + sequence: ports.get(first.invocation).sequence, + status: "returned", + output: "late", + }); + f.send({ + kind: "port-result", + ...second, + sequence: ports.get(second.invocation).sequence, + status: "returned", + output: "survived", + }); + const result = await f.receive(); + assert.equal(result.invocation, second.invocation); + assert.equal(result.output, "survived"); + assert.equal(f.io.exitCode, 0); + } finally { + await f.close(); + } +}); + +test("A to B to A uses reserved nested capacity while both root lanes are saturated", async () => { + const source = registry( + "if(request.input==='leaf')return 'done'; return await channel({portId:'next',action:{kind:'state',primitive:'read'},input:request.input});", + ); + const a = await fixture(source, { maxActive: 1, maxNested: 1, maxQueued: 1 }); + const b = await fixture(source, { maxActive: 1, maxNested: 1, maxQueued: 1 }); + const budget = (depth) => ({ root: "root-chain", depth, deadlineMs: 2000, calls: 20 }); + try { + const parent = a.start("B", budget(1)), + aPort = await a.receive(); + const neighbor = b.start("held", { ...budget(1), root: "other-root" }); + await b.receive(); + const child = b.start("A", budget(2)), + bPort = await b.receive(); + const leaf = a.start("leaf", budget(3)); + const leafResult = await a.receive(); + assert.equal(leafResult.invocation, leaf.invocation); + assert.equal(leafResult.output, "done"); + b.send({ kind: "port-result", ...child, sequence: bPort.sequence, status: "returned", output: leafResult.output }); + const childResult = await b.receive(); + a.send({ + kind: "port-result", + ...parent, + sequence: aPort.sequence, + status: "returned", + output: childResult.output, + }); + assert.equal((await a.receive()).output, "done"); + b.send({ kind: "cancel", ...neighbor }); + assert.equal((await b.receive()).kind, "cancelled"); + } finally { + await a.close(); + await b.close(); + } +}); + +test("the remaining root wall budget overrides a longer adapter deadline", async () => { + const f = await fixture( + registry("return await channel({portId:'held',action:{kind:'state',primitive:'read'},input:null});"), + { deadlineMs: 2000 }, + ); + try { + const started = performance.now(); + f.start(null, { root: "short-root", depth: 2, deadlineMs: 150, calls: 10 }); + await f.receive(); + const result = await f.receive(); + assert.equal(result.error.code, "RESTRICTED_DEADLINE"); + assert.ok(performance.now() - started < 1000); + } finally { + await f.close(); + } +}); + test("real engine returns results and resets all globals between frames", async () => { const f = await fixture( registry( @@ -222,6 +317,10 @@ test("initialization and regex loops are killed, then a fresh frame can start", f.start(); const result = await f.receive(); assert.equal(result.error.code, "RESTRICTED_DEADLINE"); + assert.ok( + f.usage.some((value) => value.cpuMicroseconds >= 50000), + "External watchdog did not account spinning engine CPU", + ); } } finally { await f.close(); @@ -268,7 +367,7 @@ test("cancellation during asynchronous preparation leaves no running frame", asy test("bounded FIFO rejects overload without losing active or queued frames", async () => { const f = await fixture( registry("return await channel({portId:'read',action:{kind:'state',primitive:'read'},input:request.input});"), - { maxQueued: 2 }, + { maxQueued: 2, maxActive: 1 }, ); try { const first = f.start(1), @@ -307,7 +406,7 @@ test("bounded FIFO rejects overload without losing active or queued frames", asy test("queue wait consumes the deadline and cancelled work never reaches its ports", async () => { const f = await fixture( registry("return await channel({portId:'read',action:{kind:'state',primitive:'read'},input:null});"), - { deadlineMs: 350 }, + { deadlineMs: 350, maxActive: 1 }, ); try { const first = f.start(); From 929fdb4a2858e8240ac24764ce899da0026aa0ea Mon Sep 17 00:00:00 2001 From: "Timothy J. Aveni" Date: Wed, 7 Oct 2026 15:59:46 -0700 Subject: [PATCH 2/4] Reuse restricted engines with fresh invocation state and revocable ports --- restricted-execution/service.mjs | 326 +++++++++++++++++--------- restricted-execution/service.test.mjs | 147 ++++++++++++ restricted-execution/worker.mjs | 17 +- 3 files changed, 377 insertions(+), 113 deletions(-) diff --git a/restricted-execution/service.mjs b/restricted-execution/service.mjs index b1ea480..aeb7fa5 100644 --- a/restricted-execution/service.mjs +++ b/restricted-execution/service.mjs @@ -1,4 +1,5 @@ import { spawn } from "node:child_process"; +import { randomUUID } from "node:crypto"; import { createServer, request as httpRequest } from "node:http"; import { mkdtemp, readFile, writeFile, rm, open } from "node:fs/promises"; import { constants } from "node:fs"; @@ -34,7 +35,8 @@ async function boundedSource(path, maximum, code) { /** Trusted stdio adapter inside one workspace VM. Engine processes never own * coordinator credentials. The supervising service must impose VM/OS limits. - * Per-frame process replacement prevents retained globals/handles crossing calls. */ + * Each engine runs one fresh Worker per invocation. Host tokens expire before + * reuse; cancellation and watchdogs destroy the affected engine, not neighbors. */ export async function serveRestrictedRegistry(config, io = process) { if ( !object(config) || @@ -68,6 +70,7 @@ export async function serveRestrictedRegistry(config, io = process) { maxQueued: config.maxQueued ?? 8, maxActive: config.maxActive ?? 2, maxNested: config.maxNested ?? 4, + engineIdleMs: config.engineIdleMs ?? 30000, }; if ( !Number.isSafeInteger(limits.maxBytes) || @@ -87,7 +90,10 @@ export async function serveRestrictedRegistry(config, io = process) { limits.maxActive > 16 || !Number.isSafeInteger(limits.maxNested) || limits.maxNested < 1 || - limits.maxNested > 16 + limits.maxNested > 16 || + !Number.isSafeInteger(limits.engineIdleMs) || + limits.engineIdleMs < 1 || + limits.engineIdleMs > 60000 ) throw fixedError("INVALID_RESTRICTED_LIMITS"); const packageSource = await boundedSource(config.module, 1024 * 1024, "RESTRICTED_SOURCE_LIMIT"); @@ -95,7 +101,10 @@ export async function serveRestrictedRegistry(config, io = process) { const sdk = []; for (const m of config.sdk) sdk.push({ ...m, source: await boundedSource(m.path, 2 * 1024 * 1024, "RESTRICTED_SDK_LIMIT") }); - const active = new Map(); + const active = new Map(), + engines = new Set(), + idleEngines = []; + const metrics = { starts: 0, reused: 0, completed: 0, peakRSSBytes: 0 }; let buffered = Buffer.alloc(0), stopped = false; const queued = []; @@ -121,7 +130,9 @@ export async function serveRestrictedRegistry(config, io = process) { throw error; } const fields = text.slice(text.lastIndexOf(")") + 2).split(" "); - const cpuMicroseconds = (Number(fields[11]) + Number(fields[12])) * 10000; + const total = (Number(fields[11]) + Number(fields[12])) * 10000; + const cpuMicroseconds = total - (call.cpuBaseline ?? 0); + if (call.engine) call.engine.cpuTotal = total; if (!Number.isSafeInteger(cpuMicroseconds) || cpuMicroseconds < (call.cpuMicroseconds ?? 0)) throw fixedError("RESTRICTED_CPU_ACCOUNTING"); if (cpuMicroseconds !== (call.cpuMicroseconds ?? 0)) { @@ -158,17 +169,23 @@ export async function serveRestrictedRegistry(config, io = process) { } catch (error) { usageFailure = error; } - if (call.child?.pid) { - try { - process.kill(-call.child.pid, "SIGKILL"); - } catch (error) { - if (error.code !== "ESRCH") throw error; - } - await call.exited; + const engine = call.engine; + if (engine) { + // Revoke the frame token before returning an engine to the idle pool. + // A delayed child request can never acquire the next call's authority. + if (engine.current === call) engine.current = null; + if (!stopped && reason.code === "INVOCATION_ENDED" && !usageFailure && !engine.destroying) { + const status = await readFile(`/proc/${engine.child.pid}/status`, "utf8").catch(() => ""); + const rss = Number(status.match(/^VmRSS:\s+(\d+)/m)?.[1] ?? Infinity) * 1024; + metrics.peakRSSBytes = Math.max(metrics.peakRSSBytes, rss); + // Bound retained engine heaps as well as lifetime. Hard aggregate + // physical memory remains the supervising process/cgroup's contract. + if (rss <= 128 * 1024 * 1024) { + idleEngines.push(engine); + engine.idleTimer = setTimeout(() => void destroyEngine(engine).catch(fail), limits.engineIdleMs); + } else await destroyEngine(engine); + } else await destroyEngine(engine); } - call.server?.closeAllConnections(); - if (call.server) await new Promise((resolve) => call.server.close(resolve)); - if (call.directory) await rm(call.directory, { recursive: true, force: true }); active.delete(call.message.invocation); if (usageFailure) throw usageFailure; })(); @@ -180,6 +197,7 @@ export async function serveRestrictedRegistry(config, io = process) { io.stdin.destroy(); for (const call of [...queued]) await retire(call, error); await Promise.all([...active.values()].map((call) => retire(call, error))); + await Promise.all([...engines].map(destroyEngine)); io.stdout.end(); io.exitCode = 1; }; @@ -193,6 +211,148 @@ export async function serveRestrictedRegistry(config, io = process) { } return JSON.parse(new TextDecoder("utf-8", { fatal: true }).decode(Buffer.concat(chunks))); }; + const destroyEngine = async (engine) => { + if (engine.destroying) return engine.destroying; + engine.destroying = (async () => { + clearTimeout(engine.idleTimer); + const index = idleEngines.indexOf(engine); + if (index >= 0) idleEngines.splice(index, 1); + engine.current = null; + if (engine.child?.pid) { + try { + process.kill(-engine.child.pid, "SIGKILL"); + } catch (error) { + if (error.code !== "ESRCH") throw error; + } + await engine.exited; + } + engine.server?.closeAllConnections(); + if (engine.server) await new Promise((resolve) => engine.server.close(resolve)); + if (engine.directory) await rm(engine.directory, { recursive: true, force: true }); + engines.delete(engine); + })(); + return engine.destroying; + }; + const prepareEngine = async (call) => { + const cached = idleEngines.pop(); + if (cached) { + clearTimeout(cached.idleTimer); + cached.current = call; + call.engine = cached; + call.child = cached.child; + call.cpuBaseline = cached.cpuTotal ?? 0; + metrics.reused++; + watchCPU(call); + return; + } + const engine = { current: call, cpuTotal: 0 }; + engines.add(engine); + call.engine = engine; + engine.directory = await mkdtemp(join(tmpdir(), "qx-engine-")); + if (!call.open) return; + const broker = join(engine.directory, "broker.sock"); + engine.socket = join(engine.directory, "engine.sock"); + engine.server = createServer(async (request, response) => { + const call = engine.current; + const message = call?.message; + try { + if (!call?.open || request.method !== "POST" || request.url !== "/port/" + call.token) + throw fixedError("EXPIRED_CONTEXT"); + const port = await body(request); + if ( + !call.open || + !object(port) || + Object.keys(port).some((key) => !["portId", "action", "input", "key"].includes(key)) || + typeof port.portId !== "string" || + !object(port.action) || + !Object.hasOwn(port, "input") + ) + throw fixedError("INVALID_RESTRICTED_PORT"); + if ( + ++call.sequence > Math.min(limits.maxPortCalls, message.budget?.calls ?? limits.maxPortCalls) || + call.pending.size >= limits.maxPortCalls + ) + throw fixedError("RESTRICTED_PORT_LIMIT"); + const sequence = call.sequence; + const result = new Promise((resolve, reject) => call.pending.set(sequence, { resolve, reject })); + void result.catch(() => {}); + send({ + kind: "port", + invocation: message.invocation, + frame: message.frame, + sequence, + portId: port.portId, + action: { ...port.action, ...(port.key === undefined ? {} : { key: port.key }) }, + input: port.input, + }); + const value = await result; + if (!call.open) throw fixedError("EXPIRED_CONTEXT"); + const encoded = JSON.stringify(value); + if (Buffer.byteLength(encoded) > limits.maxBytes) throw fixedError("RESTRICTED_MESSAGE_LIMIT"); + response.writeHead(200, { "content-type": "application/json" }); + response.end(encoded); + } catch { + response.writeHead(409); + response.end('{"status":"failed","error":{"code":"RESTRICTED_PORT_FAILED"}}'); + } + }); + + await new Promise((resolve, reject) => { + engine.server.once("error", reject); + engine.server.listen(broker, resolve); + }); + const modules = { "entry.mjs": { js: workerSource }, "package.mjs": { js: packageSource } }; + for (const module of sdk) modules[module.name] = { js: module.source }; + const setting = { registryExport: config.registryExport, exports: config.exports, maxBytes: limits.maxBytes }; + // The parent contains immutable checked source only. It never imports the + // authored module or keeps a child stub, request, result or frame handle. + const parent = `const modules=${JSON.stringify(modules)}, setting=${JSON.stringify(setting)}; + export default {async fetch(request,env) { + const input=await request.json(); + const child=env.LOADER.load({compatibilityDate:"2026-10-05",mainModule:"entry.mjs",modules, + globalOutbound:null,env:{BROKER:env.BROKER,CONFIG:{...setting,frameToken:input.frameToken}}}); + const response=await child.getEntrypoint().fetch(new Request("http://invocation/",{ + method:"POST",body:JSON.stringify(input.message)})); + const body=await response.arrayBuffer(); + return new Response(body,{status:response.status,headers:response.headers}); + }};`; + await writeFile(join(engine.directory, "parent.mjs"), parent); + const configuration = `using Workerd = import "/workerd/workerd.capnp"; const config :Workerd.Config = ( + services=[(name="main",worker=(modules=[(name="parent.mjs",esModule=embed "parent.mjs")], + compatibilityDate="2026-10-05",compatibilityFlags=["experimental"],globalOutbound="deny", + bindings=[(name="LOADER",workerLoader=()),(name="BROKER",service="broker")])), + (name="broker",external=(address=${JSON.stringify("unix:" + broker)},http=())), + (name="deny",network=(allow=[]))], + sockets=[(name="frame",address=${JSON.stringify("unix:" + engine.socket)},http=(),service="main")]);`; + const file = join(engine.directory, "config.capnp"); + await writeFile(file, configuration); + if (!call.open) return; + engine.child = spawn(config.workerd, ["serve", "--experimental", file], { + detached: true, + env: { TZ: "UTC", LANG: "C.UTF-8" }, + cwd: engine.directory, + stdio: ["ignore", "ignore", "pipe"], + }); + metrics.starts++; + engine.exited = new Promise((resolve) => { + engine.child.once("exit", resolve); + engine.child.once("error", resolve); + }); + call.child = engine.child; + watchCPU(call); + let diagnostics = 0; + engine.child.stderr.on("data", (chunk) => { + if ((diagnostics += chunk.length) > 65536) { + if (engine.current?.open) engine.current.terminate("RESTRICTED_DIAGNOSTIC_LIMIT"); + else void destroyEngine(engine).catch(fail); + } + }); + engine.child.once("error", () => engine.current?.terminate("RESTRICTED_START_FAILED")); + void engine.exited.then(() => { + if (engine.current?.open) engine.current.terminate("RESTRICTED_ENGINE_EXITED"); + else void destroyEngine(engine).catch(fail); + }); + }; const available = (call) => [...active.values()].filter((entry) => entry.nested === call.nested).length < (call.nested ? limits.maxNested : limits.maxActive); @@ -207,12 +367,21 @@ export async function serveRestrictedRegistry(config, io = process) { if (reply.kind === "cancelled") call.cancellationReply = reply; return (call.finished ??= (async () => { await retire(call, reason); + metrics.lastInvocation = { + queueMs: (call.startedAt ?? performance.now()) - call.receivedAt, + preparationMs: call.preparationMs ?? 0, + roundtripMs: call.roundtripMs ?? 0, + initializationMs: call.initializationMs ?? null, + executionMs: call.executionMs ?? null, + totalMs: performance.now() - call.receivedAt, + }; send(call.cancellationReply ?? reply); pump(); })()); }; const invoke = async (call) => { const { message } = call; + call.startedAt = performance.now(); active.set(message.invocation, call); const terminate = (code) => { if (!call.open) return; @@ -224,100 +393,12 @@ export async function serveRestrictedRegistry(config, io = process) { }; call.terminate = terminate; try { - let engine; - call.preparing = (async () => { - call.directory = await mkdtemp(join(tmpdir(), "qx-frame-")); - if (!call.open) return; - const broker = join(call.directory, "broker.sock"); - engine = join(call.directory, "engine.sock"); - call.server = createServer(async (request, response) => { - try { - if (!call.open || request.method !== "POST" || request.url !== "/port") throw fixedError("EXPIRED_CONTEXT"); - const port = await body(request); - if ( - !call.open || - !object(port) || - Object.keys(port).some((key) => !["portId", "action", "input", "key"].includes(key)) || - typeof port.portId !== "string" || - !object(port.action) || - !Object.hasOwn(port, "input") - ) - throw fixedError("INVALID_RESTRICTED_PORT"); - if ( - ++call.sequence > Math.min(limits.maxPortCalls, message.budget?.calls ?? limits.maxPortCalls) || - call.pending.size >= limits.maxPortCalls - ) - throw fixedError("RESTRICTED_PORT_LIMIT"); - const sequence = call.sequence; - const result = new Promise((resolve, reject) => call.pending.set(sequence, { resolve, reject })); - void result.catch(() => {}); - send({ - kind: "port", - invocation: message.invocation, - frame: message.frame, - sequence, - portId: port.portId, - action: { ...port.action, ...(port.key === undefined ? {} : { key: port.key }) }, - input: port.input, - }); - const value = await result; - if (!call.open) throw fixedError("EXPIRED_CONTEXT"); - const encoded = JSON.stringify(value); - if (Buffer.byteLength(encoded) > limits.maxBytes) throw fixedError("RESTRICTED_MESSAGE_LIMIT"); - response.writeHead(200, { "content-type": "application/json" }); - response.end(encoded); - } catch { - response.writeHead(409); - response.end('{"status":"failed","error":{"code":"RESTRICTED_PORT_FAILED"}}'); - } - }); - await new Promise((resolve, reject) => { - call.server.once("error", reject); - call.server.listen(broker, resolve); - }); - await writeFile(join(call.directory, "package.mjs"), packageSource); - await writeFile(join(call.directory, "entry.mjs"), workerSource); - const modules = [ - { name: "entry.mjs", esModule: 'embed "entry.mjs"' }, - { name: "package.mjs", esModule: 'embed "package.mjs"' }, - ]; - for (const [index, module] of sdk.entries()) { - const file = `sdk-${index}.mjs`; - await writeFile(join(call.directory, file), module.source); - modules.push({ name: module.name, esModule: "embed " + JSON.stringify(file) }); - } - const setting = { registryExport: config.registryExport, exports: config.exports, maxBytes: limits.maxBytes }; - const configuration = `using Workerd = import "/workerd/workerd.capnp"; const config :Workerd.Config = ( - services = [(name="main", worker=(modules=[${modules.map((m) => `(name=${JSON.stringify(m.name)}, esModule=${m.esModule})`).join(",")}], - compatibilityDate="2026-10-05", globalOutbound="deny", - bindings=[(name="BROKER", service="broker"),(name="CONFIG",json=${JSON.stringify(JSON.stringify(setting))})])), - (name="broker",external=(address=${JSON.stringify("unix:" + broker)},http=())), (name="deny",network=(allow=[]))], - sockets=[(name="frame",address=${JSON.stringify("unix:" + engine)},http=(),service="main")]);`; - const file = join(call.directory, "config.capnp"); - await writeFile(file, configuration); - if (!call.open) return; - call.child = spawn(config.workerd, ["serve", file], { - detached: true, - env: { TZ: "UTC", LANG: "C.UTF-8" }, - cwd: call.directory, - stdio: ["ignore", "ignore", "pipe"], - }); - call.exited = new Promise((resolve) => { - call.child.once("exit", resolve); - call.child.once("error", resolve); - }); - watchCPU(call); - let diagnostics = 0; - call.child.stderr.on("data", (chunk) => { - if ((diagnostics += chunk.length) > 65536) terminate("RESTRICTED_DIAGNOSTIC_LIMIT"); - }); - call.child.once("error", () => terminate("RESTRICTED_START_FAILED")); - void call.exited.then(() => { - if (call.open) terminate("RESTRICTED_ENGINE_EXITED"); - }); - })(); + call.token = randomUUID(); + call.preparing = prepareEngine(call); await call.preparing; + call.preparationMs = performance.now() - call.startedAt; if (!call.open) return; + const roundtripAt = performance.now(); const result = await new Promise((resolve, reject) => { const attempt = () => { if (!call.open) { @@ -325,10 +406,17 @@ export async function serveRestrictedRegistry(config, io = process) { return; } const request = httpRequest( - { socketPath: engine, path: "/", method: "POST", agent: false }, + { socketPath: call.engine.socket, path: "/", method: "POST", agent: false }, async (response) => { try { const result = await body(response); + for (const [header, key] of [ + ["x-quixos-initialization-ms", "initializationMs"], + ["x-quixos-execution-ms", "executionMs"], + ]) { + const value = Number(response.headers[header]); + if (Number.isFinite(value) && value >= 0 && value <= limits.deadlineMs) call[key] = value; + } if (response.statusCode !== 200) throw fixedError("RESTRICTED_PACKAGE_FAILED"); resolve(result); } catch (error) { @@ -340,10 +428,11 @@ export async function serveRestrictedRegistry(config, io = process) { if (call.open && ["ENOENT", "ECONNREFUSED"].includes(error.code)) setTimeout(attempt, 5); else reject(error); }); - request.end(JSON.stringify(message)); + request.end(JSON.stringify({ message, frameToken: call.token })); }; attempt(); }); + call.roundtripMs = performance.now() - roundtripAt; if (!call.open) return; if ( !object(result) || @@ -367,6 +456,7 @@ export async function serveRestrictedRegistry(config, io = process) { error: { code: result.error.code, message: result.error.code }, }; await finish(call, fixedError("INVOCATION_ENDED"), reply); + metrics.completed++; } catch (error) { if (call.open) { const code = /^RESTRICTED_[A-Z_]+$/.test(error.code ?? "") ? error.code : "RESTRICTED_PACKAGE_FAILED"; @@ -412,7 +502,14 @@ export async function serveRestrictedRegistry(config, io = process) { throw fixedError("INVALID_RESTRICTED_BUDGET"); if (active.has(message.invocation) || queued.some((call) => call.message.invocation === message.invocation)) throw fixedError("RESTRICTED_DUPLICATE_INVOCATION"); - const call = { message, open: true, pending: new Map(), sequence: 0, nested: (message.budget?.depth ?? 0) > 1 }; + const call = { + message, + receivedAt: performance.now(), + open: true, + pending: new Map(), + sequence: 0, + nested: (message.budget?.depth ?? 0) > 1, + }; // Descendants cannot wait behind ancestors that hold all nested capacity. // Reject excess fan-out/depth rather than deadlock an admitted call tree. if (!available(call) && (call.nested || queued.length >= limits.maxQueued)) { @@ -494,7 +591,16 @@ export async function serveRestrictedRegistry(config, io = process) { io.stdin.on("error", (error) => void fail(error)); io.stdout.on("error", (error) => void fail(error)); send({ kind: "ready", exports: config.exports }); - return { close: () => fail(fixedError("SERVICE_SHUTDOWN")) }; + return { + close: () => fail(fixedError("SERVICE_SHUTDOWN")), + stats: () => ({ + ...metrics, + active: active.size, + queued: queued.length, + engines: engines.size, + idle: idleEngines.length, + }), + }; } if (process.argv[1] === new URL(import.meta.url).pathname) { diff --git a/restricted-execution/service.test.mjs b/restricted-execution/service.test.mjs index e127ae8..70716d3 100644 --- a/restricted-execution/service.test.mjs +++ b/restricted-execution/service.test.mjs @@ -77,6 +77,7 @@ async function fixture(source, options = {}) { send, start, usage, + stats: service.stats, async close() { await service.close(); await rm(directory, { recursive: true, force: true }); @@ -424,3 +425,149 @@ test("queue wait consumes the deadline and cancelled work never reaches its port await f.close(); } }); + +test("warm engines create fresh transitive module state, closures and built-ins", async () => { + const directory = await mkdtemp(join(tmpdir(), "qx-transitive-")); + const helper = join(directory, "helper.mjs"); + await writeFile( + helper, + "let value=0, retained; export function touch(channel){const old=retained; retained=channel; return {count:++value, captured:!!old};}", + ); + const f = await fixture( + `import {touch} from 'helper.mjs'; let count=0; + export const registry={exports:['test'],async invoke(request,channel){ + const before={object:Object.prototype.retained??null,array:Array.prototype.retained??null, + map:Map.prototype.retained??null, global:globalThis.retained??null}; + Object.prototype.retained=request.input; Array.prototype.retained=request.input; + Map.prototype.retained=request.input; globalThis.retained=request.input; + return {count:++count, helper:touch(channel), before}; + }};`, + { sdk: [{ name: "helper.mjs", path: helper }] }, + ); + try { + for (const value of ["private graph A", "private graph B", "private graph C"]) { + f.start(value); + assert.deepEqual((await f.receive()).output, { + count: 1, + helper: { count: 1, captured: false }, + before: { object: null, array: null, map: null, global: null }, + }); + } + assert.equal(f.stats().starts, 1, "Qualification must exercise actual engine reuse"); + assert.equal(f.stats().reused, 2); + assert.equal(f.stats().idle, 1); + } finally { + await f.close(); + await rm(directory, { recursive: true, force: true }); + } +}); + +test("an expired frame token cannot borrow a later invocation's broker", async () => { + const directory = await mkdtemp(join(tmpdir(), "qx-token-")); + const worker = join(directory, "token-test.mjs"); + // Trusted test entrypoint deliberately replays the old token. Authored + // registries never receive CONFIG or BROKER directly. + await writeFile( + worker, + `export default {async fetch(request,env) { + const message=await request.json(); + const output=message.input ? (await env.BROKER.fetch('http://frame/port/'+message.input,{ + method:'POST',body:JSON.stringify({portId:'read',action:{kind:'state',primitive:'read'},input:null})})).status + : env.CONFIG.frameToken; + return Response.json({status:'returned',output}); + }};`, + ); + const f = await fixture(registry("return null;"), { worker }); + try { + f.start(); + const token = (await f.receive()).output; + assert.match(token, /^[a-f0-9-]{36}$/); + f.start(token); + assert.equal((await f.receive()).output, 409); + assert.equal(f.stats().starts, 1); + assert.equal(f.stats().reused, 1); + } finally { + await f.close(); + await rm(directory, { recursive: true, force: true }); + } +}); + +test("idle engines evict without a graph poll and restart with fresh state", async () => { + const f = await fixture(registry("return request.input;"), { engineIdleMs: 30 }); + try { + f.start("first"); + assert.equal((await f.receive()).output, "first"); + await new Promise((resolve) => setTimeout(resolve, 100)); + assert.equal(f.stats().engines, 0); + f.start("second"); + assert.equal((await f.receive()).output, "second"); + assert.equal(f.stats().starts, 2); + } finally { + await f.close(); + } +}); + +for (const loop of [ + "while(true) await Promise.resolve();", + "const memory=[]; for(let n=0;n<128;n++) memory.push(new Uint8Array(256*1024)); while(true) await Promise.resolve(memory);", +]) + test("await/allocation pressure is fenced without delaying a neighboring engine: " + loop.slice(0, 28), async () => { + const f = await fixture(registry(`if(request.input==='spin'){${loop}} return request.input;`), { + deadlineMs: 1500, + }); + try { + const stuck = f.start("spin", { root: "stuck", depth: 1, deadlineMs: 350, calls: 10 }); + const before = performance.now(); + const healthy = f.start("healthy"); + const replies = [await f.receive(), await f.receive()]; + assert.equal(replies.find((r) => r.invocation === healthy.invocation).output, "healthy"); + assert.equal(replies.find((r) => r.invocation === stuck.invocation).error.code, "RESTRICTED_DEADLINE"); + assert.equal(replies[0].invocation, healthy.invocation, "Healthy work must finish before the stuck deadline"); + assert.ok(performance.now() - before < 1200); + f.start("after fault"); + assert.equal((await f.receive()).output, "after fault"); + assert.equal(f.stats().starts, 2, "The healthy engine survives its neighbor's termination"); + } finally { + await f.close(); + } + }); + +test("detached microtask loops cannot escape the invocation wall deadline", async () => { + const f = await fixture( + registry("void(async()=>{while(true)await Promise.resolve();})(); return 'nominal return';"), + { deadlineMs: 200 }, + ); + try { + const before = performance.now(); + f.start(); + assert.equal((await f.receive()).error.code, "RESTRICTED_DEADLINE"); + assert.ok(performance.now() - before < 1000); + assert.equal(f.stats().engines, 0); + } finally { + await f.close(); + } +}); + +test("detached digest callbacks cannot use a completed frame or contaminate reuse", async () => { + const f = await fixture( + registry(` + void(async()=>{ + for(let n=0;n<100;n++)await crypto.subtle.digest('SHA-256',new Uint8Array(32768)); + await channel({portId:'late',action:{kind:'state',primitive:'read'},input:request.input}); + })().catch(()=>{}); + return request.input; + `), + ); + try { + f.start("private A"); + assert.equal((await f.receive()).output, "private A"); + await new Promise((resolve) => setTimeout(resolve, 75)); + f.start("private B"); + const reply = await f.receive(); + assert.equal(reply.status, "returned", "No late graph port may appear after completion"); + assert.equal(reply.output, "private B"); + assert.equal(f.stats().starts, 1); + } finally { + await f.close(); + } +}); diff --git a/restricted-execution/worker.mjs b/restricted-execution/worker.mjs index a2e21b6..3239dfd 100644 --- a/restricted-execution/worker.mjs +++ b/restricted-execution/worker.mjs @@ -1,5 +1,7 @@ // Platform entrypoint, evaluated before loading the authored registry. A fresh -// engine process owns exactly one frame. BROKER grants only that frame's ports. +// Worker owns exactly one frame. The parent engine is reusable; BROKER's host +// token grants only this frame and expires before the next invocation. +const monotonic = performance.now.bind(performance); const deny = () => { throw Error("RESTRICTED_AMBIENT_ACCESS"); }; @@ -48,7 +50,9 @@ export default { if (used) return new Response("FRAME_EXPIRED", { status: 409 }); used = true; const input = await request.json(); + const loading = monotonic(); const source = await import("./package.mjs"); + const initializationMs = monotonic() - loading; const registry = source[env.CONFIG.registryExport]; if ( !registry || @@ -59,7 +63,7 @@ export default { let open = true; const channel = async (value) => { if (!open) throw Error("EXPIRED_CONTEXT"); - const response = await env.BROKER.fetch("http://frame/port", { + const response = await env.BROKER.fetch("http://frame/port/" + env.CONFIG.frameToken, { method: "POST", body: JSON.stringify(value), }); @@ -71,6 +75,7 @@ export default { }); return result.output; }; + const running = monotonic(); try { const output = await registry.invoke( { exportId: input.exportId, workspace: input.workspace, receiver: input.receiver, input: input.input }, @@ -78,7 +83,13 @@ export default { ); const body = JSON.stringify({ status: "returned", output }); if (new TextEncoder().encode(body).length > env.CONFIG.maxBytes) throw Error("RESTRICTED_OUTPUT_LIMIT"); - return new Response(body, { headers: { "content-type": "application/json" } }); + return new Response(body, { + headers: { + "content-type": "application/json", + "x-quixos-initialization-ms": String(initializationMs), + "x-quixos-execution-ms": String(monotonic() - running), + }, + }); } catch (error) { // Only authored/codec rejection codes cross the engine boundary. Engine // diagnostics stay private to the supervisor, never become protocol data. From d6e66b537f29c33b0390680bd4bda36529159d60 Mon Sep 17 00:00:00 2001 From: "Timothy J. Aveni" Date: Wed, 7 Oct 2026 16:14:15 -0700 Subject: [PATCH 3/4] Pool host-approved restricted code with scoped and fair graph execution --- restricted-execution/service.mjs | 76 +++++++++++++++++++++++---- restricted-execution/service.test.mjs | 27 +++++++++- 2 files changed, 91 insertions(+), 12 deletions(-) diff --git a/restricted-execution/service.mjs b/restricted-execution/service.mjs index aeb7fa5..39eedcd 100644 --- a/restricted-execution/service.mjs +++ b/restricted-execution/service.mjs @@ -14,6 +14,46 @@ const absolute = (value) => value.startsWith("/") && !/[\s\u0000-\u001f]/.test(value) && !value.split("/").includes(".."); + +/** Host-wide engine reservations. Only trusted adapters hold these callbacks; + * workers cannot alter admission or evict another invocation's engine. */ +export class RestrictedEngineBudget { + #leases = new Set(); + constructor(limit = 8, reserve = async () => ({})) { + if (!Number.isSafeInteger(limit) || limit < 1 || limit > 64) throw fixedError("INVALID_ENGINE_BUDGET"); + this.limit = limit; + this.reserve = reserve; + } + async acquire(idle, evict) { + while (this.#leases.size >= this.limit) { + const victim = [...this.#leases].find((entry) => entry.idle()); + if (!victim) throw fixedError("RESTRICTED_ENGINE_LIMIT"); + await victim.evict(); + } + const lease = { idle, evict }; + this.#leases.add(lease); + try { + const resource = await this.reserve(); + let releasing; + const release = () => + (releasing ??= (async () => { + try { + await resource.release?.(); + } finally { + this.#leases.delete(lease); + } + })()); + release.attach = resource.attach; + return release; + } catch (error) { + this.#leases.delete(lease); + throw error; + } + } + stats() { + return { engines: this.#leases.size, limit: this.limit }; + } +} async function boundedSource(path, maximum, code) { const handle = await open(path, constants.O_RDONLY | constants.O_NOFOLLOW | constants.O_NONBLOCK); try { @@ -101,12 +141,14 @@ export async function serveRestrictedRegistry(config, io = process) { const sdk = []; for (const m of config.sdk) sdk.push({ ...m, source: await boundedSource(m.path, 2 * 1024 * 1024, "RESTRICTED_SDK_LIMIT") }); + const engineBudget = config.engineBudget ?? new RestrictedEngineBudget(limits.maxActive + limits.maxNested); const active = new Map(), engines = new Set(), idleEngines = []; const metrics = { starts: 0, reused: 0, completed: 0, peakRSSBytes: 0 }; let buffered = Buffer.alloc(0), - stopped = false; + stopped = false, + failure; const queued = []; const send = (value) => { if (stopped) return; @@ -191,15 +233,17 @@ export async function serveRestrictedRegistry(config, io = process) { })(); return call.retiring; }; - const fail = async (error) => { - if (stopped) return; + const fail = (error) => { + if (failure) return failure; stopped = true; - io.stdin.destroy(); - for (const call of [...queued]) await retire(call, error); - await Promise.all([...active.values()].map((call) => retire(call, error))); - await Promise.all([...engines].map(destroyEngine)); - io.stdout.end(); - io.exitCode = 1; + return (failure = (async () => { + io.stdin.destroy(); + for (const call of [...queued]) await retire(call, error); + await Promise.all([...active.values()].map((call) => retire(call, error))); + await Promise.all([...engines].map(destroyEngine)); + io.stdout.end(); + io.exitCode = 1; + })()); }; const body = async (stream) => { const chunks = []; @@ -228,8 +272,12 @@ export async function serveRestrictedRegistry(config, io = process) { } engine.server?.closeAllConnections(); if (engine.server) await new Promise((resolve) => engine.server.close(resolve)); - if (engine.directory) await rm(engine.directory, { recursive: true, force: true }); - engines.delete(engine); + try { + if (engine.directory) await rm(engine.directory, { recursive: true, force: true }); + } finally { + engines.delete(engine); + await engine.release?.(); + } })(); return engine.destroying; }; @@ -248,6 +296,11 @@ export async function serveRestrictedRegistry(config, io = process) { const engine = { current: call, cpuTotal: 0 }; engines.add(engine); call.engine = engine; + engine.release = await engineBudget.acquire( + () => !engine.current && !engine.destroying, + () => destroyEngine(engine), + ); + if (!call.open) return; engine.directory = await mkdtemp(join(tmpdir(), "qx-engine-")); if (!call.open) return; const broker = join(engine.directory, "broker.sock"); @@ -339,6 +392,7 @@ export async function serveRestrictedRegistry(config, io = process) { engine.child.once("error", resolve); }); call.child = engine.child; + await engine.release.attach?.(engine.child.pid); watchCPU(call); let diagnostics = 0; engine.child.stderr.on("data", (chunk) => { diff --git a/restricted-execution/service.test.mjs b/restricted-execution/service.test.mjs index 70716d3..a3c03f5 100644 --- a/restricted-execution/service.test.mjs +++ b/restricted-execution/service.test.mjs @@ -5,7 +5,7 @@ import { mkdtemp, writeFile, rm } from "node:fs/promises"; import { tmpdir } from "node:os"; import { join } from "node:path"; import { fileURLToPath } from "node:url"; -import { serveRestrictedRegistry } from "./service.mjs"; +import { serveRestrictedRegistry, RestrictedEngineBudget } from "./service.mjs"; const workerd = process.env.QUIXOS_TEST_WORKERD; if (!workerd) throw Error("Set QUIXOS_TEST_WORKERD to the pinned engine"); @@ -571,3 +571,28 @@ test("detached digest callbacks cannot use a completed frame or contaminate reus await f.close(); } }); + +test("the shared engine budget evicts idle code families and rejects busy oversubscription", async () => { + const engineBudget = new RestrictedEngineBudget(1); + const a = await fixture(registry("return request.input;"), { engineBudget }); + const b = await fixture( + registry("return await channel({portId:'read',action:{kind:'state',primitive:'read'},input:null});"), + { engineBudget }, + ); + try { + a.start("first family"); + assert.equal((await a.receive()).output, "first family"); + const busy = b.start(); + const port = await b.receive(); + assert.equal(a.stats().engines, 0, "Idle foreign code is evicted before allocating another engine"); + assert.equal(engineBudget.stats().engines, 1); + a.start("overload"); + assert.equal((await a.receive()).error.code, "RESTRICTED_ENGINE_LIMIT"); + assert.equal(engineBudget.stats().engines, 1); + b.send({ kind: "port-result", ...busy, sequence: port.sequence, status: "returned", output: "survived" }); + assert.equal((await b.receive()).output, "survived"); + } finally { + await Promise.all([a.close(), b.close()]); + } + assert.equal(engineBudget.stats().engines, 0); +}); From dbd159be7087f3e5585a54f5c83fe92603f52c11 Mon Sep 17 00:00:00 2001 From: "Timothy J. Aveni" Date: Wed, 7 Oct 2026 16:52:20 -0700 Subject: [PATCH 4/4] Keep checked browser runtime versions independent of authority upgrades --- restricted-execution/workerd.nix | 5 +++-- 1 file changed, 3 insertions(+), 2 deletions(-) diff --git a/restricted-execution/workerd.nix b/restricted-execution/workerd.nix index 087b9ef..35a1014 100644 --- a/restricted-execution/workerd.nix +++ b/restricted-execution/workerd.nix @@ -1,5 +1,6 @@ # Pinned upstream engine. This binary supplies an import/capability boundary; -# production execution additionally requires a workspace VM and OS limits. +# production execution additionally requires OS limits and platform approval +# for shared placement, or a workspace VM for unreviewed application code. { pkgs }: let version = "1.20261005.1"; @@ -37,7 +38,7 @@ pkgs.stdenvNoCC.mkDerivation { "$out/bin/workerd" --version | grep -Fx 'workerd 2026-10-05' ''; meta = { - description = "Pinned workerd engine for VM-scoped Quixos restricted execution"; + description = "Pinned workerd engine for Quixos restricted execution"; homepage = "https://github.com/cloudflare/workerd"; license = pkgs.lib.licenses.asl20; platforms = builtins.attrNames platforms;