770 lines
28 KiB
JavaScript
770 lines
28 KiB
JavaScript
import test from "node:test";
|
|
import assert from "node:assert/strict";
|
|
import { PassThrough } from "node:stream";
|
|
import { mkdtemp, writeFile, rm, mkdir, readdir, lstat } from "node:fs/promises";
|
|
import { tmpdir } from "node:os";
|
|
import { join, dirname } from "node:path";
|
|
import { fileURLToPath } from "node:url";
|
|
import { serveRestrictedRegistry, RestrictedEngineBudget } from "./service.mjs";
|
|
import { socketDirectory, checkedSocketPath } from "../socket-address.mjs";
|
|
|
|
const workerd = process.env.QUIXOS_TEST_WORKERD;
|
|
if (!workerd) throw Error("Set QUIXOS_TEST_WORKERD to the pinned engine");
|
|
const worker = fileURLToPath(new URL("./worker.mjs", import.meta.url));
|
|
const abi = "quixos-transaction-ports-v1";
|
|
let next = 0;
|
|
async function fixture(source, options = {}) {
|
|
const directory = await mkdtemp(join(tmpdir(), "qx-restricted-test-"));
|
|
const module = join(directory, "module.mjs");
|
|
await writeFile(module, source);
|
|
const io = { stdin: new PassThrough(), stdout: new PassThrough(), exitCode: 0 };
|
|
const waiting = [],
|
|
queued = [];
|
|
const usage = [];
|
|
let buffer = "";
|
|
io.stdout.on("data", (chunk) => {
|
|
buffer += chunk;
|
|
let index;
|
|
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);
|
|
}
|
|
});
|
|
const receive = async () => {
|
|
if (queued.length) return queued.shift();
|
|
let timer;
|
|
try {
|
|
return await Promise.race([
|
|
new Promise((resolve) => waiting.push(resolve)),
|
|
new Promise((_, reject) => {
|
|
timer = setTimeout(() => reject(Error("TEST_MESSAGE_TIMEOUT")), 5000);
|
|
}),
|
|
]);
|
|
} finally {
|
|
clearTimeout(timer);
|
|
}
|
|
};
|
|
const service = await serveRestrictedRegistry(
|
|
{
|
|
workerd,
|
|
worker,
|
|
module,
|
|
sandbox: process.env.QUIXOS_TEST_SANDBOX,
|
|
registryExport: "registry",
|
|
exports: ["test"],
|
|
sdk: [],
|
|
...options,
|
|
},
|
|
io,
|
|
);
|
|
assert.deepEqual(await receive(), { abi, kind: "ready", exports: ["test"] });
|
|
const send = (value) => io.stdin.write(JSON.stringify({ abi, ...value }) + "\n");
|
|
const start = (input = null, budget) => {
|
|
const invocation = "invoke-" + ++next,
|
|
frame = "frame-" + next;
|
|
send({
|
|
kind: "invoke",
|
|
invocation,
|
|
frame,
|
|
exportId: "test",
|
|
workspace: "workspace:restricted",
|
|
receiver: null,
|
|
input,
|
|
effect: "snapshot",
|
|
...(budget ? { budget } : {}),
|
|
});
|
|
return { invocation, frame };
|
|
};
|
|
return {
|
|
io,
|
|
receive,
|
|
send,
|
|
start,
|
|
usage,
|
|
stats: service.stats,
|
|
async close() {
|
|
await service.close();
|
|
await rm(directory, { recursive: true, force: true });
|
|
},
|
|
};
|
|
}
|
|
const registry = (code) => `export const registry={exports:['test'],async invoke(request,channel){${code}}};`;
|
|
|
|
test("real engine sockets and retirement are independent of deep Unicode TMPDIR", async () => {
|
|
const root = await mkdtemp(join(tmpdir(), "restricted-deep-"));
|
|
const deep = join(root, "保留された状態".repeat(12));
|
|
await mkdir(deep);
|
|
const oldTmp = process.env.TMPDIR;
|
|
const socketRoot = dirname(socketDirectory({ kind: "qualification-probe" }));
|
|
const names = async () =>
|
|
readdir(socketRoot).catch((error) => {
|
|
if (error.code !== "ENOENT") throw error;
|
|
return [];
|
|
});
|
|
const before = new Set(await names());
|
|
let f, namespace;
|
|
process.env.TMPDIR = deep;
|
|
try {
|
|
assert.ok(Buffer.byteLength(join(deep, "qx-engine-XXXXXX", "engine.sock")) > 100);
|
|
f = await fixture(registry("return request.input;"));
|
|
f.start("deep runtime");
|
|
const result = await f.receive();
|
|
assert.equal(result.status, "returned");
|
|
assert.equal(result.output, "deep runtime");
|
|
assert.equal(f.stats().starts, 1);
|
|
const created = (await names()).filter((name) => !before.has(name));
|
|
assert.equal(created.length, 1);
|
|
namespace = join(socketRoot, created[0]);
|
|
const engines = await readdir(namespace);
|
|
assert.equal(engines.length, 1);
|
|
for (const name of ["broker.sock", "engine.sock"]) {
|
|
const endpoint = checkedSocketPath(join(namespace, engines[0], name));
|
|
assert.ok((await lstat(endpoint)).isSocket());
|
|
}
|
|
} finally {
|
|
try {
|
|
await f?.close();
|
|
} finally {
|
|
if (oldTmp === undefined) delete process.env.TMPDIR;
|
|
else process.env.TMPDIR = oldTmp;
|
|
}
|
|
}
|
|
await assert.rejects(lstat(namespace), { code: "ENOENT" });
|
|
assert.ok((await lstat(deep)).isDirectory());
|
|
await rm(root, { recursive: true, force: true });
|
|
});
|
|
|
|
test("uncertain OS cleanup cannot release engine capacity", async () => {
|
|
const budget = new RestrictedEngineBudget(1, async () => ({
|
|
async release() {
|
|
throw Error("kernel cleanup not confirmed");
|
|
},
|
|
}));
|
|
const release = await budget.acquire(
|
|
() => false,
|
|
async () => {},
|
|
);
|
|
await assert.rejects(release(), /cleanup not confirmed/);
|
|
assert.equal(budget.stats().engines, 1);
|
|
await assert.rejects(
|
|
budget.acquire(
|
|
() => false,
|
|
async () => {},
|
|
),
|
|
{ code: "RESTRICTED_ENGINE_LIMIT" },
|
|
);
|
|
});
|
|
|
|
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(
|
|
"globalThis.retained=(globalThis.retained??0)+1; return {count:globalThis.retained,text:new Intl.Segmenter('zh',{granularity:'word'}).segment('雨后的书店').containing(3).segment};",
|
|
),
|
|
);
|
|
try {
|
|
for (let n = 0; n < 3; n++) {
|
|
const call = f.start();
|
|
const result = await f.receive();
|
|
assert.equal(result.invocation, call.invocation);
|
|
assert.equal(result.status, "returned");
|
|
assert.equal(result.output.count, 1);
|
|
assert.equal(typeof result.output.text, "string");
|
|
}
|
|
} finally {
|
|
await f.close();
|
|
}
|
|
});
|
|
test("captured frame is the sole source of nested port identities", async () => {
|
|
const f = await fixture(
|
|
registry("return await channel({portId:'read',action:{kind:'state',primitive:'read'},input:null});"),
|
|
);
|
|
try {
|
|
const call = f.start();
|
|
const port = await f.receive();
|
|
assert.equal(port.kind, "port");
|
|
assert.equal(port.frame, call.frame);
|
|
assert.equal(port.invocation, call.invocation);
|
|
f.send({ kind: "port-result", ...call, sequence: port.sequence, status: "returned", output: 42 });
|
|
const result = await f.receive();
|
|
assert.equal(result.status, "returned");
|
|
assert.equal(result.output, 42);
|
|
} finally {
|
|
await f.close();
|
|
}
|
|
});
|
|
test("authored serialization cannot forge result routing or ABI fields", async () => {
|
|
const f = await fixture(
|
|
registry(
|
|
"Object.prototype.toJSON=function(){return this.status==='returned'?{status:'returned',output:42,invocation:'forged',abi:'forged'}:this;};return 42;",
|
|
),
|
|
);
|
|
try {
|
|
const call = f.start();
|
|
const result = await f.receive();
|
|
assert.equal(result.invocation, call.invocation);
|
|
assert.equal(result.abi, abi);
|
|
assert.equal(result.status, "failed");
|
|
assert.equal(result.error.code, "RESTRICTED_PACKAGE_FAILED");
|
|
} finally {
|
|
await f.close();
|
|
}
|
|
});
|
|
test("the checked portable SDK keeps codecs and nested graph ports inside the frame", async () => {
|
|
const sdk = process.env.QUIXOS_TEST_SDK;
|
|
assert.ok(sdk, "Set QUIXOS_TEST_SDK to the checked portable SDK");
|
|
const C = await import(sdk),
|
|
workspace = C.workspaceId("workspace:restricted");
|
|
const integer = { kind: "scalar", name: "int64" },
|
|
unit = { kind: "builtin", name: "unit" };
|
|
const declarations = [
|
|
{
|
|
id: "test",
|
|
name: "test",
|
|
effect: "snapshot",
|
|
receiver: null,
|
|
input: integer,
|
|
output: integer,
|
|
methods: [
|
|
{
|
|
name: "get",
|
|
portName: "count",
|
|
portId: "count-read",
|
|
action: { kind: "state", primitive: "read" },
|
|
input: unit,
|
|
output: integer,
|
|
keyed: false,
|
|
},
|
|
],
|
|
},
|
|
];
|
|
const source = `import {definePortableRegistry} from 'qx:core'; export const registry=definePortableRegistry(${JSON.stringify(declarations)},{test:async ctx=>ctx.input+await ctx.ports.count.get()});`;
|
|
const f = await fixture(source, { sdk: [{ name: "qx:core", path: sdk }] });
|
|
try {
|
|
const call = f.start(C.encodeArgument(integer, 2n, { workspace }));
|
|
const port = await f.receive();
|
|
assert.equal(port.kind, "port");
|
|
assert.equal(C.decodeArgument(unit, port.input, { workspace }), null);
|
|
f.send({
|
|
kind: "port-result",
|
|
...call,
|
|
sequence: port.sequence,
|
|
status: "returned",
|
|
output: C.encodeArgument(integer, 4n, { workspace }),
|
|
});
|
|
const result = await f.receive();
|
|
assert.equal(result.status, "returned");
|
|
assert.equal(C.decodeArgument(integer, result.output, { workspace }), 6n);
|
|
} finally {
|
|
await f.close();
|
|
}
|
|
});
|
|
test("checked handlers hash canonical values without ambient capabilities", async () => {
|
|
const sdk = process.env.QUIXOS_TEST_SDK;
|
|
const C = await import(sdk);
|
|
const workspace = C.workspaceId("workspace:restricted");
|
|
const text = { kind: "scalar", name: "string" };
|
|
const value = {
|
|
kind: "record",
|
|
fields: {
|
|
text,
|
|
options: {
|
|
kind: "record",
|
|
fields: {
|
|
level: { kind: "scalar", name: "int32" },
|
|
enabled: { kind: "scalar", name: "bool" },
|
|
},
|
|
},
|
|
},
|
|
};
|
|
const declarations = [
|
|
{ id: "test", name: "test", effect: "snapshot", receiver: null, input: value, output: text, methods: [] },
|
|
];
|
|
const source = `import {definePortableRegistry,hashJson} from 'qx:core'; export const registry=definePortableRegistry(${JSON.stringify(declarations)},{test:ctx=>hashJson('test-content',ctx.input)});`;
|
|
const f = await fixture(source, { sdk: [{ name: "qx:core", path: sdk }] });
|
|
try {
|
|
const input = { text: "雨后的书店", options: { level: 3, enabled: true } };
|
|
let expected;
|
|
for (const candidate of [input, { options: input.options, text: input.text }, { ...input, text: "別の文" }]) {
|
|
f.start(C.encodeArgument(value, candidate, { workspace }));
|
|
const result = await f.receive();
|
|
assert.equal(result.status, "returned");
|
|
const actual = C.decodeArgument(text, result.output, { workspace });
|
|
assert.equal(actual, await C.hashJson("test-content", candidate));
|
|
if (candidate.text === input.text) {
|
|
expected ??= actual;
|
|
assert.equal(actual, expected, "Object key order must not invalidate content caches");
|
|
} else assert.notEqual(actual, expected, "Content changes must invalidate the key");
|
|
}
|
|
} finally {
|
|
await f.close();
|
|
}
|
|
});
|
|
test("ambient clock, entropy, network and native imports are denied", async () => {
|
|
for (const code of [
|
|
"Date.now()",
|
|
"new Date()",
|
|
"new Intl.DateTimeFormat().format()",
|
|
"new Intl.DateTimeFormat().formatToParts()",
|
|
"Math.random()",
|
|
"crypto.randomUUID()",
|
|
"fetch('https://example.com')",
|
|
"(await import('node:fs')).readFileSync('/etc/passwd')",
|
|
"(await import('./unlisted.mjs')).secret",
|
|
]) {
|
|
const f = await fixture(registry(`return ${code};`));
|
|
try {
|
|
f.start();
|
|
assert.equal((await f.receive()).status, "failed", code);
|
|
} finally {
|
|
await f.close();
|
|
}
|
|
}
|
|
});
|
|
test("explicit dates remain useful without an implicit clock", async () => {
|
|
const f = await fixture(registry("return new Date('2026-01-01T00:00:00Z').toISOString();"));
|
|
try {
|
|
f.start();
|
|
assert.equal((await f.receive()).output, "2026-01-01T00:00:00.000Z");
|
|
} finally {
|
|
await f.close();
|
|
}
|
|
});
|
|
test("initialization and regex loops are killed, then a fresh frame can start", async () => {
|
|
for (const source of [
|
|
"while(true){};" + registry("return 1;"),
|
|
registry("return /(a+)+$/.test('a'.repeat(100000)+'!');"),
|
|
]) {
|
|
const f = await fixture(source, { deadlineMs: 350 });
|
|
try {
|
|
for (let n = 0; n < 2; n++) {
|
|
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();
|
|
}
|
|
}
|
|
});
|
|
test("cancellation kills the entire frame before acknowledgment and rejects late port replies", async () => {
|
|
const f = await fixture(
|
|
registry("return await channel({portId:'read',action:{kind:'state',primitive:'read'},input:null});"),
|
|
);
|
|
try {
|
|
const call = f.start();
|
|
const port = await f.receive();
|
|
f.send({ kind: "cancel", ...call });
|
|
const cancelled = await f.receive();
|
|
assert.equal(cancelled.kind, "cancelled");
|
|
f.send({ kind: "port-result", ...call, sequence: port.sequence, status: "returned", output: "late" });
|
|
const next = f.start();
|
|
const nextPort = await f.receive();
|
|
assert.equal(nextPort.invocation, next.invocation);
|
|
f.send({ kind: "port-result", ...next, sequence: nextPort.sequence, status: "returned", output: "fresh" });
|
|
const result = await f.receive();
|
|
assert.equal(result.output, "fresh");
|
|
assert.equal(result.invocation, next.invocation);
|
|
} finally {
|
|
await f.close();
|
|
}
|
|
});
|
|
test("cancellation during asynchronous preparation leaves no running frame", async () => {
|
|
const f = await fixture(registry("return 1;"));
|
|
try {
|
|
for (let n = 0; n < 5; n++) {
|
|
const call = f.start();
|
|
f.send({ kind: "cancel", ...call });
|
|
assert.equal((await f.receive()).kind, "cancelled");
|
|
}
|
|
f.start();
|
|
assert.equal((await f.receive()).output, 1);
|
|
} finally {
|
|
await f.close();
|
|
}
|
|
});
|
|
|
|
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, maxActive: 1 },
|
|
);
|
|
try {
|
|
const first = f.start(1),
|
|
port = await f.receive();
|
|
const second = f.start(2),
|
|
third = f.start(3),
|
|
overflow = f.start(4);
|
|
const refused = await f.receive();
|
|
assert.equal(refused.invocation, overflow.invocation);
|
|
assert.equal(refused.error.code, "RESTRICTED_QUEUE_FULL");
|
|
f.send({ kind: "cancel", ...second });
|
|
assert.equal((await f.receive()).invocation, second.invocation);
|
|
f.send({ kind: "port-result", ...first, sequence: port.sequence, status: "returned", output: 1 });
|
|
assert.equal((await f.receive()).output, 1);
|
|
const nextPort = await f.receive();
|
|
assert.equal(nextPort.invocation, third.invocation);
|
|
assert.equal(nextPort.input, 3);
|
|
f.send({ kind: "port-result", ...third, sequence: nextPort.sequence, status: "returned", output: 3 });
|
|
assert.equal((await f.receive()).output, 3);
|
|
f.start(5);
|
|
const fresh = await f.receive();
|
|
f.send({
|
|
kind: "port-result",
|
|
invocation: fresh.invocation,
|
|
frame: fresh.frame,
|
|
sequence: fresh.sequence,
|
|
status: "returned",
|
|
output: 5,
|
|
});
|
|
assert.equal((await f.receive()).output, 5);
|
|
} finally {
|
|
await f.close();
|
|
}
|
|
});
|
|
|
|
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: 4000, maxActive: 1 },
|
|
);
|
|
try {
|
|
const first = f.start();
|
|
assert.equal((await f.receive()).kind, "port");
|
|
// Expire while the first call is deliberately blocked. Two equal relative
|
|
// deadlines let fast startup legitimately begin the queued call first.
|
|
const queued = f.start(null, { root: "queued-deadline", deadlineMs: 50, depth: 1, calls: 10 });
|
|
const expired = await f.receive();
|
|
assert.equal(expired.invocation, queued.invocation);
|
|
assert.equal(expired.error.code, "RESTRICTED_DEADLINE");
|
|
f.send({ kind: "cancel", ...first });
|
|
const cancelled = await f.receive();
|
|
assert.equal(cancelled.invocation, first.invocation);
|
|
assert.equal(cancelled.kind, "cancelled");
|
|
const fresh = f.start();
|
|
const port = await f.receive();
|
|
assert.equal(port.invocation, fresh.invocation);
|
|
f.send({ kind: "port-result", ...fresh, sequence: port.sequence, status: "returned", output: "fresh" });
|
|
assert.equal((await f.receive()).output, "fresh");
|
|
} finally {
|
|
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 updates = [];
|
|
const f = await fixture(registry("return request.input;"), {
|
|
engineIdleMs: 30,
|
|
onChange: () => updates.push(f.stats()),
|
|
});
|
|
try {
|
|
f.start("first");
|
|
assert.equal((await f.receive()).output, "first");
|
|
await new Promise((resolve) => setTimeout(resolve, 100));
|
|
assert.equal(f.stats().engines, 0);
|
|
assert.ok(
|
|
updates.some((s) => s.active > 0),
|
|
"Inspectors observe active invocation admission",
|
|
);
|
|
assert.ok(
|
|
updates.some((s) => s.idle > 0 && s.idleDeadline),
|
|
"Idle eviction deadline is visible",
|
|
);
|
|
assert.equal(updates.at(-1).engines, 0, "Eviction pushes an update without a graph poll");
|
|
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();
|
|
}
|
|
});
|
|
|
|
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);
|
|
});
|
|
|
|
test("engine retirement accounting keeps its admission lease busy", async () => {
|
|
const observations = [];
|
|
let idle;
|
|
const engineBudget = new RestrictedEngineBudget(1, async () => ({
|
|
usage: async () => {
|
|
observations.push(idle());
|
|
await new Promise((done) => setImmediate(done));
|
|
return { cpuMicroseconds: 0, memoryBytes: 1024 };
|
|
},
|
|
release: async () => {},
|
|
}));
|
|
const acquire = engineBudget.acquire.bind(engineBudget);
|
|
engineBudget.acquire = (isIdle, evict) => {
|
|
idle = isIdle;
|
|
return acquire(isIdle, evict);
|
|
};
|
|
const f = await fixture(registry("return request.input;"), { engineBudget });
|
|
try {
|
|
f.start("first");
|
|
assert.equal((await f.receive()).output, "first");
|
|
assert.ok(observations.length > 0);
|
|
assert.ok(
|
|
observations.every((value) => !value),
|
|
"An engine must not become evictable during asynchronous retirement accounting",
|
|
);
|
|
assert.equal(f.stats().idle, 1);
|
|
assert.equal(idle(), true);
|
|
f.start("second");
|
|
assert.equal((await f.receive()).output, "second");
|
|
assert.equal(f.stats().starts, 1);
|
|
} finally {
|
|
await f.close();
|
|
}
|
|
assert.equal(engineBudget.stats().engines, 0);
|
|
});
|