diff --git a/toybox/scripts/hello.js b/toybox/scripts/hello.js index 94321f8..3bc1e7c 100644 --- a/toybox/scripts/hello.js +++ b/toybox/scripts/hello.js @@ -12,7 +12,8 @@ const name = process.argv[2] ?? "hello"; const client = await connectClient("admin.sock", `hello-script-${process.pid}`); try { const ingress = await E(client.vatAdmin).makeIngress({ vat: name }); - const { id } = await E(client.vatAdmin).createVat(await readFile("vats/hello.js", "utf8"), { ingress }); + const { id, root } = await E(client.vatAdmin).createVat(await readFile("vats/hello.js", "utf8"), { ingress }); + await root; // its handoff, settled before the client goes console.log(`vat ${id}\nhttp://127.0.0.1:8787/?vat=${name}`); } finally { client.close(); diff --git a/toybox/supervisor.js b/toybox/supervisor.js index dc9bda3..f174173 100644 --- a/toybox/supervisor.js +++ b/toybox/supervisor.js @@ -2,7 +2,7 @@ import "@endo/init"; import { DurableObject, WorkerEntrypoint } from "cloudflare:workers"; import { RpcTarget, newWebSocketRpcSession, newWorkersRpcResponse } from "capnweb"; -import { E, Far } from "@endo/far"; +import { E, Far, passStyleOf } from "@endo/far"; import { makeClient, swissnumFromBytes } from "@endo/ocapn"; import { mergeLabels, urlLabels } from "./labels.js"; import { LOOKUP, MOUNT, MOUNT_LABELS, MOUNT_SCHEMA, OVERLAP, UNMOUNT, UNMOUNT_LABELS } from "./mounts-sql.js"; @@ -178,29 +178,136 @@ const encodeUtf8 = harden(TextEncoder.prototype.encode.bind(new TextEncoder())); */ export class VatAdmin extends DurableObject { #clients = new Map(); - /** mount key → the handler serving it (a live reference, for now) */ + /** mount key → the handler serving it: live, so lost when this object is evicted */ #handlers = new Map(); + /** mount key → resolvers waiting for a handler */ + #waiting = new Map(); + /** mount key ↔ its ingress object, made again as needed */ + #ingresses = new Map(); + #ingressKeys = new WeakMap(); + #reviving = new Map(); #ready; constructor(ctx, env) { super(ctx, env); + const { sql } = ctx.storage; + // a vat's params, encoded (see #encode); a vat absent here can't be restarted + sql.exec("CREATE TABLE IF NOT EXISTS vat_params (vat TEXT PRIMARY KEY, params TEXT)"); + // which vat was given which ingress, so a request can restart it + sql.exec("CREATE TABLE IF NOT EXISTS ingress_vat (key TEXT PRIMARY KEY, vat TEXT)"); this.#ready = this.#boot(); } + /** @param {string} key */ + #ingressFor(key) { + let ingress = this.#ingresses.get(key); + if (!ingress) { + ingress = Far("ingress", { + /** @param {{ handle: (request: object) => object }} handler */ + serve: (handler) => { + this.#handlers.set(key, handler); + for (const resolve of this.#waiting.get(key) ?? []) resolve(); + this.#waiting.delete(key); + }, + }); + this.#ingresses.set(key, ingress); + this.#ingressKeys.set(ingress, key); + } + return ingress; + } + + /** + * `params` as JSON, an ingress by its key; `null` if anything in it can't + * be made again here (a remote object of another peer, say). + */ + #encode(params) { + const keys = []; + const walk = (v) => { + const style = passStyleOf(v); + if (style === "remotable") { + const key = this.#ingressKeys.get(v); + if (!key) throw new Error("not restartable"); + keys.push(key); + return { "@ingress": key }; + } + if (style === "copyArray") return v.map(walk); + if (style === "copyRecord") return Object.fromEntries(Object.entries(v).map(([k, x]) => [k, walk(x)])); + if (["string", "number", "boolean", "null"].includes(style)) return v; + throw new Error("not restartable"); + }; + try { + return { json: JSON.stringify(walk(params)), keys }; + } catch { + return null; + } + } + + #decode(json) { + const walk = (v) => { + if (Array.isArray(v)) return v.map(walk); + if (v && typeof v === "object") { + if ("@ingress" in v) return this.#ingressFor(v["@ingress"]); + return Object.fromEntries(Object.entries(v).map(([k, x]) => [k, walk(x)])); + } + return v; + }; + return harden(walk(JSON.parse(json))); + } + + /** (Re)start a vat's code with `params`: it rebuilds its root. */ + async #startVat(vat, params) { + const vatBootstrap = await this.#enliven(vat, VAT_SWISSNUM); + return E(vatBootstrap).start(params); + } + + /** + * The handler for a mount whose handler went with an evicted instance: + * restart the vat given its ingress, and wait for it to serve again. + * + * @param {string} key + */ + #revive(key) { + let reviving = this.#reviving.get(key); + if (!reviving) { + reviving = (async () => { + const [row] = this.ctx.storage.sql + .exec("SELECT i.vat, p.params FROM ingress_vat i JOIN vat_params p USING (vat) WHERE i.key = ?", key) + .toArray(); + if (!row) return null; + const served = new Promise((resolve) => this.#waiting.set(key, [...(this.#waiting.get(key) ?? []), resolve])); + await this.#startVat(row.vat, this.#decode(row.params)); + await Promise.race([served, new Promise((resolve) => setTimeout(resolve, 5000))]); + return this.#handlers.get(key) ?? null; + })().finally(() => this.#reviving.delete(key)); + this.#reviving.set(key, reviving); + } + return reviving; + } + async #boot() { const vatAdmin = Far("vatAdmin", { /** * Create an OCapN vat running `src`; `params` (any passable, * remote objects included) reach its `buildRootObject`. * + * If `params` holds only data and ingresses, they are kept, and the + * vat is restarted with them when a request needs it after an + * eviction or a restart. + * * @param {string} src an ES module exporting `buildRootObject(vatPowers, params)` */ createVat: async (src, params = {}) => { const id = this.env.VATS.newUniqueId(); + const vat = id.toString(); await this.env.VATS.get(id).start(src, null, "ocapn"); - const vatBootstrap = await this.#enliven(id.toString(), VAT_SWISSNUM); - const root = await E(vatBootstrap).start(params); - return harden({ id: id.toString(), root }); + const encoded = this.#encode(params); + if (encoded) { + const { sql } = this.ctx.storage; + sql.exec("INSERT INTO vat_params VALUES (?, ?)", vat, encoded.json); + for (const key of encoded.keys) sql.exec("INSERT OR REPLACE INTO ingress_vat VALUES (?, ?)", key, vat); + } + const root = await this.#startVat(vat, params); + return harden({ id: vat, root }); }, /** * Reserve a mount at `labels` and return the ingress for it: whoever @@ -212,10 +319,7 @@ export class VatAdmin extends DurableObject { makeIngress: async (labels) => { const key = `ingress:${crypto.randomUUID()}`; await this.env.DIRECTORY.getByName("directory").mount(key, harden({ ...labels })); - return Far("ingress", { - /** @param {{ handle: (request: object) => object }} handler */ - serve: (handler) => void this.#handlers.set(key, handler), - }); + return this.#ingressFor(key); }, }); // workerd runs FinalizationRegistry callbacks outside any request, where @@ -266,7 +370,7 @@ export class VatAdmin extends DurableObject { */ async serveHttp(key, request) { await this.#ready; - const handler = this.#handlers.get(key); + const handler = this.#handlers.get(key) ?? (await this.#revive(key)); return handler ? E(handler).handle(harden(request)) : null; } diff --git a/toybox/test/ocapn.test.js b/toybox/test/ocapn.test.js index 4eef297..745300e 100644 --- a/toybox/test/ocapn.test.js +++ b/toybox/test/ocapn.test.js @@ -12,11 +12,8 @@ const vatSrc = (name) => readFile(join(here, "vats", name), "utf8"); let dir, port, server, client; -before(async () => { - dir = await mkdtemp(join(tmpdir(), "toybox-ocapn-")); - await mkdir(join(dir, "state")); - port = await freePort(); - server = await startWorkerd( +const startToybox = () => + startWorkerd( "config.capnp", [ "-s", `http=127.0.0.1:${port}`, "-s", `admin=unix:${join(dir, "admin.sock")}`, @@ -25,6 +22,12 @@ before(async () => { { port, host: "127.0.0.1" }, [join(dir, "admin.sock")], ); + +before(async () => { + dir = await mkdtemp(join(tmpdir(), "toybox-ocapn-")); + await mkdir(join(dir, "state")); + port = await freePort(); + server = await startToybox(); client = await connectClient(join(dir, "admin.sock"), "client-1", { verbose: !!process.env.OCAPN_LOG }); }); @@ -66,3 +69,37 @@ test("an ingress nobody serves is a 503", async () => { await E(client.vatAdmin).makeIngress({ vat: "idle" }); assert.equal((await fetch(`http://127.0.0.1:${port}/?vat=idle`)).status, 503); }); + +test("after a crash, a request restarts the vat that serves it", async () => { + // its own toybox: a crash breaks every session to it, the shared client's too + const home = await mkdtemp(join(tmpdir(), "toybox-phoenix-")); + await mkdir(join(home, "state")); + const port2 = await freePort(); + const start = () => + startWorkerd( + "config.capnp", + [ + "-s", `http=127.0.0.1:${port2}`, "-s", `admin=unix:${join(home, "admin.sock")}`, + "-d", `state=${join(home, "state")}`, "-e", `switchboard=unix:${join(home, "no-switchboard.sock")}`, + ], + { port: port2, host: "127.0.0.1" }, + [join(home, "admin.sock")], + ); + let toybox = await start(); + try { + const creator = await connectClient(join(home, "admin.sock"), "creator"); + const ingress = await E(creator.vatAdmin).makeIngress({ vat: "phoenix" }); + const { root } = await E(creator.vatAdmin).createVat(await vatSrc("hello.js"), { ingress }); + assert.equal((await fetch(`http://127.0.0.1:${port2}/?vat=phoenix`)).status, 200); + await root; // its handoff, settled before the client goes + creator.close(); + await toybox.stop(); + toybox = await start(); + const resp = await fetch(`http://127.0.0.1:${port2}/again?vat=phoenix`); + assert.equal(resp.status, 200); + assert.match(await resp.text(), /Hello, world.*at \/again/); + } finally { + await toybox.stop(); + await rm(home, { recursive: true, force: true }); + } +}); diff --git a/toybox/vat-runtime.js b/toybox/vat-runtime.js index 2f1231e..6fbc6d6 100644 --- a/toybox/vat-runtime.js +++ b/toybox/vat-runtime.js @@ -32,8 +32,9 @@ export class App extends DurableObject { let root; const vatPowers = Object.freeze({ E, Far, storage: this.ctx.storage.kv }); const bootstrap = Far("vatBootstrap", { + /** (Re)build the root: on creation, and when the vatAdmin restarts the vat. */ start: (params) => { - root ??= vat.buildRootObject(vatPowers, params); + root = vat.buildRootObject(vatPowers, params); return root; }, });