721 lines
29 KiB
JavaScript
721 lines
29 KiB
JavaScript
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";
|
|
import { tmpdir } from "node:os";
|
|
import { join, dirname } from "node:path";
|
|
|
|
const abi = "quixos-transaction-ports-v1";
|
|
const fixedError = (code) => Object.assign(Error(code), { code });
|
|
const object = (v) => !!v && typeof v === "object" && !Array.isArray(v);
|
|
const absolute = (value) =>
|
|
typeof value === "string" &&
|
|
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 () => {
|
|
await resource.release?.();
|
|
this.#leases.delete(lease);
|
|
})());
|
|
release.attach = resource.attach;
|
|
release.usage = resource.usage;
|
|
release.kill = resource.kill;
|
|
release.uid = resource.uid;
|
|
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 {
|
|
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.
|
|
* 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) ||
|
|
!absolute(config.workerd) ||
|
|
!absolute(config.worker) ||
|
|
!absolute(config.module) ||
|
|
(config.sandbox !== undefined && !absolute(config.sandbox)) ||
|
|
(config.sandboxIdentity !== undefined && typeof config.sandboxIdentity !== "boolean") ||
|
|
!/^[$A-Z_a-z][$\w]*$/.test(config.registryExport) ||
|
|
!Array.isArray(config.exports) ||
|
|
!config.exports.length ||
|
|
config.exports.length > 10000 ||
|
|
config.exports.some((x) => typeof x !== "string" || !x || x.length > 4096) ||
|
|
new Set(config.exports).size !== config.exports.length ||
|
|
!Array.isArray(config.sdk) ||
|
|
config.sdk.length > 8 ||
|
|
config.sdk.some(
|
|
(m) =>
|
|
!object(m) ||
|
|
!absolute(m.path) ||
|
|
typeof m.name !== "string" ||
|
|
!m.name ||
|
|
/[\u0000-\u001f]/.test(m.name) ||
|
|
["entry.mjs", "package.mjs"].includes(m.name),
|
|
) ||
|
|
new Set(config.sdk.map((m) => m.name)).size !== config.sdk.length
|
|
)
|
|
throw fixedError("INVALID_RESTRICTED_CONFIGURATION");
|
|
const limits = {
|
|
maxBytes: config.maxBytes ?? 1024 * 1024,
|
|
maxPortCalls: config.maxPortCalls ?? 1000,
|
|
deadlineMs: config.deadlineMs ?? 10000,
|
|
maxQueued: config.maxQueued ?? 8,
|
|
maxActive: config.maxActive ?? 2,
|
|
maxNested: config.maxNested ?? 4,
|
|
engineIdleMs: config.engineIdleMs ?? 30000,
|
|
};
|
|
if (
|
|
!Number.isSafeInteger(limits.maxBytes) ||
|
|
limits.maxBytes < 1024 ||
|
|
limits.maxBytes > 4 * 1024 * 1024 ||
|
|
!Number.isSafeInteger(limits.maxPortCalls) ||
|
|
limits.maxPortCalls < 1 ||
|
|
limits.maxPortCalls > 10000 ||
|
|
!Number.isSafeInteger(limits.deadlineMs) ||
|
|
limits.deadlineMs < 1 ||
|
|
limits.deadlineMs > 60000 ||
|
|
!Number.isSafeInteger(limits.maxQueued) ||
|
|
limits.maxQueued < 0 ||
|
|
limits.maxQueued > 64 ||
|
|
!Number.isSafeInteger(limits.maxActive) ||
|
|
limits.maxActive < 1 ||
|
|
limits.maxActive > 16 ||
|
|
!Number.isSafeInteger(limits.maxNested) ||
|
|
limits.maxNested < 1 ||
|
|
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");
|
|
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 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 };
|
|
const changed = () => {
|
|
try {
|
|
config.onChange?.();
|
|
} catch {}
|
|
};
|
|
let buffered = Buffer.alloc(0),
|
|
stopped = false,
|
|
failure;
|
|
const queued = [];
|
|
const send = (value) => {
|
|
if (stopped) return;
|
|
const line = JSON.stringify({ abi, ...value }) + "\n";
|
|
if (
|
|
Buffer.byteLength(line) > limits.maxBytes ||
|
|
io.stdout.writableLength + Buffer.byteLength(line) > limits.maxBytes * 2
|
|
)
|
|
throw fixedError("RESTRICTED_BACKPRESSURE");
|
|
io.stdout.write(line);
|
|
};
|
|
// A sandbox has namespace helpers as well as workerd. Charge the complete
|
|
// delegated cgroup when available; the qualification fallback follows the
|
|
// bounded process tree, never trusting a worker's self-reported counters.
|
|
const engineUsage = async (engine) => {
|
|
if (engine.release?.usage) return engine.release.usage();
|
|
engine.processTicks ??= new Map();
|
|
const pending = [engine.child.pid];
|
|
let memoryBytes = 0,
|
|
visited = 0;
|
|
while (pending.length) {
|
|
if (++visited > 64) throw fixedError("RESTRICTED_PROCESS_LIMIT");
|
|
const pid = pending.pop();
|
|
try {
|
|
const stat = await readFile(`/proc/${pid}/stat`, "utf8");
|
|
const fields = stat.slice(stat.lastIndexOf(")") + 2).split(" ");
|
|
engine.processTicks.set(`${pid}:${fields[19]}`, Number(fields[11]) + Number(fields[12]));
|
|
const status = await readFile(`/proc/${pid}/status`, "utf8");
|
|
memoryBytes += Number(status.match(/^VmRSS:\s+(\d+)/m)?.[1] ?? 0) * 1024;
|
|
const children = await readFile(`/proc/${pid}/task/${pid}/children`, "utf8");
|
|
pending.push(...children.trim().split(/\s+/).filter(Boolean).map(Number));
|
|
} catch (error) {
|
|
if (!["ENOENT", "ESRCH"].includes(error.code)) throw error;
|
|
}
|
|
}
|
|
return {
|
|
cpuMicroseconds: [...engine.processTicks.values()].reduce((sum, ticks) => sum + ticks * 10000, 0),
|
|
memoryBytes,
|
|
};
|
|
};
|
|
const sampleCPU = async (call) => {
|
|
if (!call.child?.pid) return;
|
|
const { cpuMicroseconds: total } = await engineUsage(call.engine);
|
|
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)) {
|
|
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);
|
|
call.pending.clear();
|
|
// 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;
|
|
}
|
|
const engine = call.engine;
|
|
if (engine) {
|
|
if (!stopped && reason.code === "INVOCATION_ENDED" && !usageFailure && !engine.destroying) {
|
|
const rss = (await engineUsage(engine)).memoryBytes;
|
|
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.
|
|
// Keep the lease busy through every await. Another artifact can evict
|
|
// an idle lease immediately; never publish an engine still retiring
|
|
// or one whose process died while its final usage read was pending.
|
|
if (rss <= 128 * 1024 * 1024 && !stopped && !engine.destroying && engine.current === call) {
|
|
// Revoke the frame token atomically with publishing idle capacity.
|
|
engine.current = null;
|
|
idleEngines.push(engine);
|
|
engine.idleUntil = Date.now() + limits.engineIdleMs;
|
|
engine.idleTimer = setTimeout(() => void destroyEngine(engine).catch(fail), limits.engineIdleMs);
|
|
} else await destroyEngine(engine);
|
|
} else await destroyEngine(engine);
|
|
}
|
|
active.delete(call.message.invocation);
|
|
if (usageFailure) throw usageFailure;
|
|
})();
|
|
return call.retiring;
|
|
};
|
|
const fail = (error) => {
|
|
if (failure) return failure;
|
|
stopped = true;
|
|
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 = [];
|
|
let bytes = 0;
|
|
for await (const chunk of stream) {
|
|
bytes += chunk.length;
|
|
if (bytes > limits.maxBytes) throw fixedError("RESTRICTED_MESSAGE_LIMIT");
|
|
chunks.push(chunk);
|
|
}
|
|
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) {
|
|
if (engine.release.kill) await engine.release.kill();
|
|
else {
|
|
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));
|
|
try {
|
|
if (engine.directory) await rm(engine.directory, { recursive: true, force: true });
|
|
} finally {
|
|
engines.delete(engine);
|
|
await engine.release?.();
|
|
changed();
|
|
}
|
|
})();
|
|
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.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");
|
|
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;
|
|
if (config.sandboxIdentity && !Number.isSafeInteger(engine.release.uid))
|
|
throw fixedError("RESTRICTED_IDENTITY_UNAVAILABLE");
|
|
engine.child = spawn(
|
|
config.sandbox ?? config.workerd,
|
|
config.sandbox
|
|
? [engine.directory, ...(config.sandboxIdentity ? [String(engine.release.uid)] : [])]
|
|
: ["serve", "--experimental", file],
|
|
{
|
|
detached: true,
|
|
env: { TZ: "UTC", LANG: "C.UTF-8" },
|
|
cwd: engine.directory,
|
|
stdio: [config.sandbox ? "pipe" : "ignore", "ignore", "pipe"],
|
|
},
|
|
);
|
|
metrics.starts++;
|
|
changed();
|
|
engine.exited = new Promise((resolve) => {
|
|
engine.child.once("exit", resolve);
|
|
engine.child.once("error", resolve);
|
|
});
|
|
engine.child.stdin?.on("error", () => engine.current?.terminate("RESTRICTED_START_FAILED"));
|
|
call.child = engine.child;
|
|
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);
|
|
} else config.diagnostics?.(chunk);
|
|
});
|
|
await engine.release.attach?.(engine.child.pid);
|
|
if (config.sandbox && call.open) engine.child.stdin.end(Buffer.from([1]));
|
|
watchCPU(call);
|
|
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);
|
|
const pump = () => {
|
|
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) => {
|
|
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,
|
|
};
|
|
metrics.lastFailure =
|
|
reply.status === "failed" && /^[A-Z0-9_]{1,128}$/.test(reply.error?.code) ? reply.error.code : null;
|
|
changed();
|
|
send(call.cancellationReply ?? reply);
|
|
pump();
|
|
})());
|
|
};
|
|
const invoke = async (call) => {
|
|
const { message } = call;
|
|
call.startedAt = performance.now();
|
|
active.set(message.invocation, call);
|
|
changed();
|
|
const terminate = (code) => {
|
|
if (!call.open) return;
|
|
void finish(call, fixedError(code), {
|
|
invocation: message.invocation,
|
|
status: "failed",
|
|
error: { code, message: code },
|
|
}).catch(fail);
|
|
};
|
|
call.terminate = terminate;
|
|
try {
|
|
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) {
|
|
reject(fixedError("EXPIRED_CONTEXT"));
|
|
return;
|
|
}
|
|
const request = httpRequest(
|
|
{ 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) {
|
|
reject(error);
|
|
}
|
|
},
|
|
);
|
|
request.once("error", (error) => {
|
|
if (call.open && ["ENOENT", "ECONNREFUSED"].includes(error.code)) setTimeout(attempt, 5);
|
|
else reject(error);
|
|
});
|
|
request.end(JSON.stringify({ message, frameToken: call.token }));
|
|
};
|
|
attempt();
|
|
});
|
|
call.roundtripMs = performance.now() - roundtripAt;
|
|
if (!call.open) return;
|
|
if (
|
|
!object(result) ||
|
|
!["returned", "failed"].includes(result.status) ||
|
|
Object.keys(result).some(
|
|
(key) => !(result.status === "returned" ? ["status", "output"] : ["status", "error"]).includes(key),
|
|
) ||
|
|
(result.status === "returned"
|
|
? !Object.hasOwn(result, "output")
|
|
: !object(result.error) || !/^[A-Z0-9_]{1,128}$/.test(result.error.code))
|
|
)
|
|
throw fixedError("RESTRICTED_PACKAGE_FAILED");
|
|
// Authored code can alter serialization prototypes. Treat even the
|
|
// platform worker's JSON as untrusted; never accept routing/ABI fields.
|
|
const reply =
|
|
result.status === "returned"
|
|
? { invocation: message.invocation, status: "returned", output: result.output }
|
|
: {
|
|
invocation: message.invocation,
|
|
status: "failed",
|
|
error: { code: result.error.code, message: result.error.code },
|
|
};
|
|
await finish(call, fixedError("INVOCATION_ENDED"), reply);
|
|
metrics.completed++;
|
|
} catch (error) {
|
|
config.diagnostics?.(Buffer.from(`RESTRICTED_HOST_ERROR:${error.code ?? error.name}\n`));
|
|
if (call.open) {
|
|
const code = /^RESTRICTED_[A-Z_]+$/.test(error.code ?? "") ? error.code : "RESTRICTED_PACKAGE_FAILED";
|
|
await finish(call, error, { invocation: message.invocation, status: "failed", error: { code, message: code } });
|
|
}
|
|
}
|
|
};
|
|
const receive = (message) => {
|
|
if (!object(message) || message.abi !== abi) throw fixedError("RESTRICTED_ABI_MISMATCH");
|
|
if (message.kind === "shutdown") {
|
|
void fail(fixedError("SERVICE_SHUTDOWN"));
|
|
return;
|
|
}
|
|
if (message.kind === "invoke") {
|
|
if (
|
|
!config.exports.includes(message.exportId) ||
|
|
!["snapshot", "transaction"].includes(message.effect) ||
|
|
typeof message.invocation !== "string" ||
|
|
!message.invocation ||
|
|
message.invocation.length > 4096 ||
|
|
typeof message.frame !== "string" ||
|
|
!message.frame ||
|
|
message.frame.length > 4096
|
|
)
|
|
throw fixedError("INVALID_RESTRICTED_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");
|
|
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)) {
|
|
send({
|
|
invocation: message.invocation,
|
|
status: "failed",
|
|
error: { code: "RESTRICTED_QUEUE_FULL", message: "RESTRICTED_QUEUE_FULL" },
|
|
});
|
|
return;
|
|
}
|
|
// 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);
|
|
},
|
|
Math.min(limits.deadlineMs, message.budget?.deadlineMs ?? limits.deadlineMs),
|
|
);
|
|
queued.push(call);
|
|
pump();
|
|
return;
|
|
}
|
|
const waiting = queued.find(
|
|
(call) => call.message.invocation === message.invocation && call.message.frame === message.frame,
|
|
);
|
|
if (waiting && message.kind === "cancel") {
|
|
void finish(waiting, fixedError("INVOCATION_CANCELLED"), {
|
|
kind: "cancelled",
|
|
invocation: message.invocation,
|
|
frame: message.frame,
|
|
}).catch(fail);
|
|
return;
|
|
}
|
|
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");
|
|
return;
|
|
}
|
|
if (message.kind === "cancel") {
|
|
void finish(call, fixedError("INVOCATION_CANCELLED"), {
|
|
kind: "cancelled",
|
|
invocation: message.invocation,
|
|
frame: message.frame,
|
|
}).catch(fail);
|
|
return;
|
|
}
|
|
if (message.kind !== "port-result" || !["returned", "failed"].includes(message.status))
|
|
throw fixedError("INVALID_RESTRICTED_RESULT");
|
|
const pending = call.pending.get(message.sequence);
|
|
if (!pending) throw fixedError("RESTRICTED_UNKNOWN_PORT");
|
|
call.pending.delete(message.sequence);
|
|
pending.resolve(message);
|
|
};
|
|
io.stdin.on("data", (chunk) => {
|
|
if (stopped) return;
|
|
try {
|
|
let offset = 0;
|
|
while (offset < chunk.length) {
|
|
const newline = chunk.indexOf(10, offset),
|
|
end = newline < 0 ? chunk.length : newline;
|
|
if (buffered.length + end - offset > limits.maxBytes) throw fixedError("RESTRICTED_MESSAGE_LIMIT");
|
|
buffered = Buffer.concat([buffered, chunk.subarray(offset, end)]);
|
|
if (newline < 0) break;
|
|
receive(JSON.parse(new TextDecoder("utf-8", { fatal: true }).decode(buffered)));
|
|
buffered = Buffer.alloc(0);
|
|
offset = newline + 1;
|
|
if (stopped) break;
|
|
}
|
|
} catch (error) {
|
|
void fail(error);
|
|
}
|
|
});
|
|
io.stdin.on("end", () => void fail(fixedError("COORDINATOR_DISCONNECTED")));
|
|
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")),
|
|
stats: () => ({
|
|
...metrics,
|
|
active: active.size,
|
|
queued: queued.length,
|
|
engines: engines.size,
|
|
idle: idleEngines.length,
|
|
idleDeadline: idleEngines.length
|
|
? new Date(Math.min(...idleEngines.map((engine) => engine.idleUntil))).toISOString()
|
|
: null,
|
|
}),
|
|
};
|
|
}
|
|
|
|
if (process.argv[1] === new URL(import.meta.url).pathname) {
|
|
const config = JSON.parse(await readFile(process.argv[2], "utf8"));
|
|
await serveRestrictedRegistry({
|
|
...config,
|
|
workerd: process.env.QUIXOS_RESTRICTED_WORKERD,
|
|
worker: process.env.QUIXOS_RESTRICTED_WORKER,
|
|
sandbox: process.env.QUIXOS_RESTRICTED_SANDBOX,
|
|
module: join(dirname(process.argv[2]), config.module),
|
|
});
|
|
}
|