Timothy J. Aveni
dec3aa475a
Show per-export placement, static eligibility blockers, bounded metadata and live pool/idle/timing diagnostics. Attach inspector updates through one private shard stream and the existing authenticated SSE hub without waking graph views or VMs. Fix packed agent preparation to use OpenSSH and the worker’s own private systemd credential.
719 lines
29 KiB
JavaScript
719 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 ?? 4000,
|
|
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) {
|
|
// 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 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.
|
|
if (rss <= 128 * 1024 * 1024) {
|
|
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),
|
|
});
|
|
}
|