Confine restricted engines before launch and preserve the reader recovery baseline
This commit is contained in:
@@ -37,13 +37,11 @@ export class RestrictedEngineBudget {
|
||||
let releasing;
|
||||
const release = () =>
|
||||
(releasing ??= (async () => {
|
||||
try {
|
||||
await resource.release?.();
|
||||
} finally {
|
||||
this.#leases.delete(lease);
|
||||
}
|
||||
await resource.release?.();
|
||||
this.#leases.delete(lease);
|
||||
})());
|
||||
release.attach = resource.attach;
|
||||
release.usage = resource.usage;
|
||||
return release;
|
||||
} catch (error) {
|
||||
this.#leases.delete(lease);
|
||||
@@ -83,6 +81,7 @@ export async function serveRestrictedRegistry(config, io = process) {
|
||||
!absolute(config.workerd) ||
|
||||
!absolute(config.worker) ||
|
||||
!absolute(config.module) ||
|
||||
(config.sandbox !== undefined && !absolute(config.sandbox)) ||
|
||||
!/^[$A-Z_a-z][$\w]*$/.test(config.registryExport) ||
|
||||
!Array.isArray(config.exports) ||
|
||||
!config.exports.length ||
|
||||
@@ -160,19 +159,38 @@ 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.
|
||||
// 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;
|
||||
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 total = (Number(fields[11]) + Number(fields[12])) * 10000;
|
||||
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))
|
||||
@@ -217,8 +235,7 @@ export async function serveRestrictedRegistry(config, io = process) {
|
||||
// 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;
|
||||
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.
|
||||
@@ -380,19 +397,25 @@ export async function serveRestrictedRegistry(config, io = process) {
|
||||
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"],
|
||||
});
|
||||
engine.child = spawn(
|
||||
config.sandbox ?? config.workerd,
|
||||
config.sandbox ? [engine.directory] : ["serve", "--experimental", file],
|
||||
{
|
||||
detached: true,
|
||||
env: { TZ: "UTC", LANG: "C.UTF-8" },
|
||||
cwd: engine.directory,
|
||||
stdio: [config.sandbox ? "pipe" : "ignore", "ignore", "pipe"],
|
||||
},
|
||||
);
|
||||
metrics.starts++;
|
||||
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;
|
||||
await engine.release.attach?.(engine.child.pid);
|
||||
if (config.sandbox && call.open) engine.child.stdin.end(Buffer.from([1]));
|
||||
watchCPU(call);
|
||||
let diagnostics = 0;
|
||||
engine.child.stderr.on("data", (chunk) => {
|
||||
@@ -663,6 +686,7 @@ if (process.argv[1] === new URL(import.meta.url).pathname) {
|
||||
...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),
|
||||
});
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user