capsite/toybox: OCapN vats; createVat takes remote objects
The admin socket now also speaks OCapN over Cap'n Web: a WebSocket there reaches the VatAdmin DO, an OCapN peer whose bootstrap (swissnum vatAdmin) offers createVat(src, params). params may hold remote objects; the vat's root comes back as one. Each OCapN vat is its own peer: SES and @endo/ocapn run in its facet (vat-runtime.js, bundled with esbuild), around the vat's own module, which exports buildRootObject(vatPowers, params). Passing a client's object to a vat, or a vat's root to the client, is an OCapN handoff, so the receiver dials the exporter. Nothing here can dial out, so the VatAdmin is also the router: a client registers its gateway under a designator, a vat's designator is its id, and every peer reaches every other through it. The netlayer is ertp-clerk's, made to work over Workers RPC stubs and plain promises as well as Cap'n Web. Not yet: sturdyrefs that survive a restart, an adminNode, and grants (egress, mount) for OCapN vats. App vats (export class App) still work beside them. Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
This commit is contained in:
1 parent
8d8900d84d
commit
9356f63c64
13 files changed
+983
-61
No files matched your search
@@ -0,0 +1,133 @@
|
||||
/**
|
||||
* @file An OCapN netlayer whose transport is an RPC session: Cap'n Web, or
|
||||
* Workers RPC, or one proxied through the other.
|
||||
*
|
||||
* Each OCapN connection is a pair of {@link ByteInbox}es, one per
|
||||
* direction, exchanged through a gateway; each `deliver` call carries
|
||||
* exactly one OCapN message, so no framing is needed. The RPC session
|
||||
* provides ordering and only-the-peer-can-inject; OCapN does the rest.
|
||||
*
|
||||
* After vattr97's ertp-clerk (`ocapn-capnweb-netlayer.ts`), with the stub
|
||||
* methods only Cap'n Web has (`onRpcBroken`) made optional.
|
||||
*
|
||||
* @see https://github.com/ocapn/ocapn/blob/main/draft-specifications/Netlayers.md
|
||||
*/
|
||||
import { RpcTarget } from "capnweb";
|
||||
|
||||
export const TRANSPORT = "capnweb";
|
||||
|
||||
/**
|
||||
* A peer's location, by designator alone: a router finds the peer.
|
||||
*
|
||||
* @param {string} designator
|
||||
*/
|
||||
export const locationOf = (designator) => ({ type: "ocapn-peer", transport: TRANSPORT, designator, hints: false });
|
||||
|
||||
/** Same as `locationToLocationId` in `@endo/ocapn`, which its exports map doesn't expose. */
|
||||
const locationId = ({ designator, transport }) => `ocapn://${designator}.${transport}`;
|
||||
|
||||
/** Receives one direction of an OCapN connection, one message per call. */
|
||||
export class ByteInbox extends RpcTarget {
|
||||
#onMessage;
|
||||
|
||||
constructor(onMessage) {
|
||||
super();
|
||||
this.#onMessage = onMessage;
|
||||
}
|
||||
|
||||
/** @param {Uint8Array} bytes */
|
||||
deliver(bytes) {
|
||||
// OCapN's syrup reader needs offset 0; a pooled Buffer view may not have it.
|
||||
this.#onMessage(bytes.byteOffset === 0 ? bytes : new Uint8Array(bytes));
|
||||
}
|
||||
}
|
||||
|
||||
const dispose = (stub) => stub?.[Symbol.dispose]?.();
|
||||
|
||||
/** Write to a remote inbox, discarding each result at once so the session retains none. */
|
||||
const makeSocket = (inbox, close) => ({
|
||||
write: (bytes) => dispose(inbox.deliver(bytes)),
|
||||
end: close,
|
||||
});
|
||||
|
||||
/**
|
||||
* A remote inbox reached through a gateway that is only a promise (a
|
||||
* local async call, not a pipelining stub): writes queue in order behind it.
|
||||
*/
|
||||
const deferredInbox = (gatewayP, myInbox) => {
|
||||
const inboxP = Promise.resolve(gatewayP).then((gateway) => gateway.connect(myInbox));
|
||||
return {
|
||||
deliver: (bytes) => void inboxP.then((inbox) => dispose(inbox.deliver(bytes))),
|
||||
onRpcBroken: (cb) => void inboxP.then((inbox) => inbox.onRpcBroken?.(cb), cb),
|
||||
};
|
||||
};
|
||||
|
||||
/**
|
||||
* Make an OCapN netlayer for `client.registerNetlayer`.
|
||||
*
|
||||
* @param {object} opts
|
||||
* @param {string} opts.designator this peer's designator
|
||||
* @param {(designator: string) => any} opts.openGateway the gateway of
|
||||
* the peer with this designator: a stub or pipelining promise
|
||||
* (`openGateway(d).connect(...)`), or a plain promise for one
|
||||
* @returns a netlayer maker; the netlayer has a `gateway` that accepts
|
||||
* incoming connections
|
||||
*/
|
||||
export const makeRpcNetLayer = ({ designator, openGateway }) => {
|
||||
const location = locationOf(designator);
|
||||
const connections = new Set();
|
||||
|
||||
return (handlers, logger) => {
|
||||
const track = (connection, inbox) => {
|
||||
connections.add(connection);
|
||||
inbox.onRpcBroken?.((reason) => {
|
||||
if (connections.delete(connection)) logger.log("rpc netlayer: connection broken", reason);
|
||||
handlers.handleConnectionClose(connection, reason);
|
||||
});
|
||||
};
|
||||
|
||||
const netlayer = {
|
||||
location,
|
||||
locationId: locationId(location),
|
||||
connect: (peer) => {
|
||||
const gateway = openGateway(peer.designator);
|
||||
let connection;
|
||||
const myInbox = new ByteInbox((bytes) => handlers.handleMessageData(connection, bytes));
|
||||
// Pipelined: writes before the peer answers queue in order.
|
||||
const peerInbox =
|
||||
typeof gateway.connect === "function" ? gateway.connect(myInbox) : deferredInbox(gateway, myInbox);
|
||||
connection = handlers.makeConnection(
|
||||
netlayer,
|
||||
true,
|
||||
makeSocket(peerInbox, () => {
|
||||
connections.delete(connection);
|
||||
dispose(gateway);
|
||||
}),
|
||||
);
|
||||
track(connection, peerInbox);
|
||||
return connection;
|
||||
},
|
||||
shutdown: () => {
|
||||
for (const connection of [...connections]) connection.end();
|
||||
},
|
||||
};
|
||||
|
||||
class Gateway extends RpcTarget {
|
||||
connect(peerInbox) {
|
||||
const peer = peerInbox.dup?.() ?? peerInbox;
|
||||
const connection = handlers.makeConnection(
|
||||
netlayer,
|
||||
false,
|
||||
makeSocket(peer, () => {
|
||||
connections.delete(connection);
|
||||
dispose(peer);
|
||||
}),
|
||||
);
|
||||
track(connection, peer);
|
||||
return new ByteInbox((bytes) => handlers.handleMessageData(connection, bytes));
|
||||
}
|
||||
}
|
||||
|
||||
return Object.assign(netlayer, { gateway: new Gateway() });
|
||||
};
|
||||
};
|
||||
Reference in new issue
Block a user