Do Not Merge: Stream multiplexing epic

packages/errors/src/index.test.ts

@@ -7,6 +7,8 @@ describe('index', () => {
     expect(Object.keys(indexModule).sort()).toStrictEqual([
       'ErrorCode',
       'ErrorSentinel',
+      'MarshaledErrorStruct',
+      'MarshaledOcapErrorStruct',
       'StreamReadError',
       'VatAlreadyExistsError',
       'VatCapTpConnectionExistsError',

packages/errors/src/index.ts

@@ -5,7 +5,12 @@ export { VatAlreadyExistsError } from './errors/VatAlreadyExistsError.js';
 export { VatDeletedError } from './errors/VatDeletedError.js';
 export { VatNotFoundError } from './errors/VatNotFoundError.js';
 export { StreamReadError } from './errors/StreamReadError.js';
-export { ErrorCode, ErrorSentinel } from './constants.js';
+export {
+  ErrorCode,
+  ErrorSentinel,
+  MarshaledErrorStruct,
+  MarshaledOcapErrorStruct,
+} from './constants.js';
 export { toError } from './utils/toError.js';
 export { isOcapError } from './utils/isOcapError.js';
 export { marshalError } from './marshal/marshalError.js';

packages/extension/src/background.ts

@@ -1,6 +1,6 @@
 import type { Json } from '@metamask/utils';
-import { ClusterCommandMethod, isClusterCommandReply } from '@ocap/kernel';
-import type { ClusterCommand } from '@ocap/kernel';
+import { KernelCommandMethod, isKernelCommandReply } from '@ocap/kernel';
+import type { KernelCommand } from '@ocap/kernel';
 import { ChromeRuntimeTarget, ChromeRuntimeDuplexStream } from '@ocap/streams';
 
 const OFFSCREEN_DOCUMENT_PATH = '/offscreen.html';
@@ -31,7 +31,7 @@ async function main(): Promise<void> {
    *
    * @param command - The command to send.
    */
-  const sendClusterCommand = async (command: ClusterCommand): Promise<void> => {
+  const sendClusterCommand = async (command: KernelCommand): Promise<void> => {
     await offscreenStream.write(command);
   };
 
@@ -40,21 +40,21 @@ async function main(): Promise<void> {
     capTpCall: {
       value: async (method: string, params: Json[]) =>
         sendClusterCommand({
-          method: ClusterCommandMethod.CapTpCall,
+          method: KernelCommandMethod.capTpCall,
           params: { method, params },
         }),
     },
     evaluate: {
       value: async (source: string) =>
         sendClusterCommand({
-          method: ClusterCommandMethod.Evaluate,
+          method: KernelCommandMethod.evaluate,
           params: source,
         }),
     },
     ping: {
       value: async () =>
         sendClusterCommand({
-          method: ClusterCommandMethod.Ping,
+          method: KernelCommandMethod.ping,
           params: null,
         }),
     },
@@ -64,14 +64,14 @@ async function main(): Promise<void> {
     kvGet: {
       value: async (key: string) =>
         sendClusterCommand({
-          method: ClusterCommandMethod.KVGet,
+          method: KernelCommandMethod.kvGet,
           params: key,
         }),
     },
     kvSet: {
       value: async (key: string, value: string) =>
         sendClusterCommand({
-          method: ClusterCommandMethod.KVSet,
+          method: KernelCommandMethod.kvSet,
           params: { key, value },
         }),
     },
@@ -81,24 +81,24 @@ async function main(): Promise<void> {
   // With this we can click the extension action button to wake up the service worker.
   chrome.action.onClicked.addListener(() => {
     sendClusterCommand({
-      method: ClusterCommandMethod.Ping,
+      method: KernelCommandMethod.ping,
       params: null,
     }).catch(console.error);
   });
 
   // Handle replies from the offscreen document
   for await (const message of offscreenStream) {
-    if (!isClusterCommandReply(message)) {
+    if (!isKernelCommandReply(message)) {
       console.error('Background received unexpected message', message);
       continue;
     }
 
     switch (message.method) {
-      case ClusterCommandMethod.Evaluate:
-      case ClusterCommandMethod.CapTpCall:
-      case ClusterCommandMethod.Ping:
-      case ClusterCommandMethod.KVGet:
-      case ClusterCommandMethod.KVSet:
+      case KernelCommandMethod.evaluate:
+      case KernelCommandMethod.capTpCall:
+      case KernelCommandMethod.ping:
+      case KernelCommandMethod.kvGet:
+      case KernelCommandMethod.kvSet:
         console.log(message.params);
         break;
       default:

packages/extension/src/iframe.ts

@@ -1,29 +1,26 @@
 import { makeExo } from '@endo/exo';
 import { M } from '@endo/patterns';
 import { Supervisor } from '@ocap/kernel';
-import type { StreamEnvelope, StreamEnvelopeReply } from '@ocap/kernel';
-import { MessagePortDuplexStream, receiveMessagePort } from '@ocap/streams';
+import { MessagePortMultiplexer, receiveMessagePort } from '@ocap/streams';
 
 main().catch(console.error);
 
 /**
  * The main function for the iframe.
  */
 async function main(): Promise<void> {
-  const stream = await receiveMessagePort(
+  const multiplexer = await receiveMessagePort(
     (listener) => addEventListener('message', listener),
     (listener) => removeEventListener('message', listener),
-  ).then(async (port) =>
-    MessagePortDuplexStream.make<StreamEnvelope, StreamEnvelopeReply>(port),
-  );
+  ).then(async (port) => new MessagePortMultiplexer(port));
 
   const bootstrap = makeExo(
     'TheGreatFrangooly',
     M.interface('TheGreatFrangooly', {}, { defaultGuards: 'passable' }),
     { whatIsTheGreatFrangooly: () => 'Crowned with Chaos' },
   );
 
-  const supervisor = new Supervisor({ id: 'iframe', stream, bootstrap });
+  const supervisor = new Supervisor({ id: 'iframe', multiplexer, bootstrap });
 
   console.log(supervisor.evaluate('["Hello", "world!"].join(" ");'));
 }

packages/extension/src/kernel/VatWorkerClient.test.ts

@@ -40,8 +40,8 @@ describe('ExtensionVatWorkerClient', () => {
 
   it.each`
     method
-    ${VatWorkerServiceCommandMethod.Launch}
-    ${VatWorkerServiceCommandMethod.Terminate}
+    ${VatWorkerServiceCommandMethod.launch}
+    ${VatWorkerServiceCommandMethod.terminate}
   `(
     "calls logger.error when receiving a $method reply it wasn't waiting for",
     async ({ method }) => {
@@ -63,15 +63,15 @@ describe('ExtensionVatWorkerClient', () => {
     },
   );
 
-  it(`calls logger.error when receiving a ${VatWorkerServiceCommandMethod.Launch} reply without a port`, async () => {
+  it(`calls logger.error when receiving a ${VatWorkerServiceCommandMethod.launch} reply without a port`, async () => {
     const errorSpy = vi.spyOn(clientLogger, 'error');
     const vatId: VatId = 'v0';
     // eslint-disable-next-line @typescript-eslint/no-floating-promises
     client.launch(vatId);
     const reply = {
       id: 'm1',
       payload: {
-        method: VatWorkerServiceCommandMethod.Launch,
+        method: VatWorkerServiceCommandMethod.launch,
         params: { vatId: 'v0' },
       },
     };

packages/extension/src/kernel/VatWorkerClient.ts

@@ -7,14 +7,12 @@ import {
   isVatWorkerServiceCommandReply,
 } from '@ocap/kernel';
 import type {
-  StreamEnvelope,
-  StreamEnvelopeReply,
   VatWorkerService,
   VatId,
   VatWorkerServiceCommand,
 } from '@ocap/kernel';
-import type { DuplexStream } from '@ocap/streams';
-import { MessagePortDuplexStream } from '@ocap/streams';
+import type { DuplexStream, MultiplexEnvelope } from '@ocap/streams';
+import { isMultiplexEnvelope, MessagePortDuplexStream } from '@ocap/streams';
 import type { Logger } from '@ocap/utils';
 import { makeCounter, makeHandledCallback, makeLogger } from '@ocap/utils';
 
@@ -77,23 +75,23 @@ export class ExtensionVatWorkerClient implements VatWorkerService {
 
   async launch(
     vatId: VatId,
-  ): Promise<DuplexStream<StreamEnvelopeReply, StreamEnvelope>> {
+  ): Promise<DuplexStream<MultiplexEnvelope, MultiplexEnvelope>> {
     return this.#sendMessage({
-      method: VatWorkerServiceCommandMethod.Launch,
+      method: VatWorkerServiceCommandMethod.launch,
       params: { vatId },
     });
   }
 
   async terminate(vatId: VatId): Promise<undefined> {
     return this.#sendMessage({
-      method: VatWorkerServiceCommandMethod.Terminate,
+      method: VatWorkerServiceCommandMethod.terminate,
       params: { vatId },
     });
   }
 
   async terminateAll(): Promise<void> {
     return this.#sendMessage({
-      method: VatWorkerServiceCommandMethod.TerminateAll,
+      method: VatWorkerServiceCommandMethod.terminateAll,
       params: null,
     });
   }
@@ -122,19 +120,20 @@ export class ExtensionVatWorkerClient implements VatWorkerService {
     }
 
     switch (method) {
-      case VatWorkerServiceCommandMethod.Launch:
+      case VatWorkerServiceCommandMethod.launch:
         if (!port) {
           this.#logger.error('Expected a port with message reply', event);
           return;
         }
         promise.resolve(
-          MessagePortDuplexStream.make<StreamEnvelope, StreamEnvelopeReply>(
+          new MessagePortDuplexStream<MultiplexEnvelope, MultiplexEnvelope>(
             port,
+            isMultiplexEnvelope,
           ),
         );
         break;
-      case VatWorkerServiceCommandMethod.Terminate:
-      case VatWorkerServiceCommandMethod.TerminateAll:
+      case VatWorkerServiceCommandMethod.terminate:
+      case VatWorkerServiceCommandMethod.terminateAll:
         // If we were caching streams on the client this would be a good place
         // to remove them.
         promise.resolve(undefined);

packages/extension/src/kernel/VatWorkerServer.test.ts

@@ -76,14 +76,14 @@ describe('ExtensionVatWorkerServer', () => {
       clientPort.postMessage({
         id: 'm0',
         payload: {
-          method: VatWorkerServiceCommandMethod.Launch,
+          method: VatWorkerServiceCommandMethod.launch,
           params: { vatId },
         },
       });
       clientPort.postMessage({
         id: 'm1',
         payload: {
-          method: VatWorkerServiceCommandMethod.TerminateAll,
+          method: VatWorkerServiceCommandMethod.terminateAll,
           params: null,
         },
       });
@@ -92,7 +92,7 @@ describe('ExtensionVatWorkerServer', () => {
 
       expect(errorSpy).toHaveBeenCalledOnce();
       expect(errorSpy.mock.lastCall?.[0]).toBe(
-        `Error handling ${VatWorkerServiceCommandMethod.TerminateAll} for vatId ${vatId}`,
+        `Error handling ${VatWorkerServiceCommandMethod.terminateAll} for vatId ${vatId}`,
       );
       expect(errorSpy.mock.lastCall?.[1]).toBe(vatNotFoundError);
     });

packages/extension/src/kernel/VatWorkerServer.ts

@@ -86,17 +86,17 @@ export class ExtensionVatWorkerServer {
     };
 
     switch (method) {
-      case VatWorkerServiceCommandMethod.Launch:
+      case VatWorkerServiceCommandMethod.launch:
         await this.#launch(params.vatId)
           .then((port) => this.#postMessage({ id, payload }, [port]))
           .catch(async (error) => handleError(error, params.vatId));
         break;
-      case VatWorkerServiceCommandMethod.Terminate:
+      case VatWorkerServiceCommandMethod.terminate:
         await this.#terminate(params.vatId)
           .then(() => this.#postMessage({ id, payload }))
           .catch(async (error) => handleError(error, params.vatId));
         break;
-      case VatWorkerServiceCommandMethod.TerminateAll:
+      case VatWorkerServiceCommandMethod.terminateAll:
         await Promise.all(
           Array.from(this.#vatWorkers.keys()).map(async (vatId) =>
             this.#terminate(vatId).catch((error) => handleError(error, vatId)),

packages/extension/src/kernel/kernel-worker.ts

@@ -65,7 +65,7 @@ async function runVatLifecycle(
   const vatToPing = vats[Math.floor(Math.random() * vats.length)] as VatId;
   console.time(`Ping Vat "${vatToPing}"`);
   await kernel.sendMessage(vatToPing, {
-    method: VatCommandMethod.Ping,
+    method: VatCommandMethod.ping,
     params: null,
   });
   console.timeEnd(`Ping Vat "${vatToPing}"`);

packages/kernel/src/Kernel.test.ts

@@ -1,7 +1,11 @@
 import '@ocap/shims/endoify';
 
 import { VatAlreadyExistsError, VatNotFoundError } from '@ocap/errors';
-import type { MessagePortDuplexStream, DuplexStream } from '@ocap/streams';
+import type {
+  MessagePortDuplexStream,
+  DuplexStream,
+  MultiplexEnvelope,
+} from '@ocap/streams';
 import type { MockInstance } from 'vitest';
 import { describe, it, expect, vi, beforeEach } from 'vitest';
 
@@ -12,7 +16,6 @@ import type {
   KernelCommandReply,
   VatCommand,
 } from './messages/index.js';
-import type { StreamEnvelope, StreamEnvelopeReply } from './stream-envelope.js';
 import type { VatId, VatWorkerService } from './types.js';
 import { Vat } from './Vat.js';
 import { makeMapKVStore } from '../test/storage.js';
@@ -45,9 +48,7 @@ describe('Kernel', () => {
 
     launchWorkerMock = vi
       .spyOn(mockWorkerService, 'launch')
-      .mockResolvedValue(
-        {} as DuplexStream<StreamEnvelopeReply, StreamEnvelope>,
-      );
+      .mockResolvedValue({} as DuplexStream<MultiplexEnvelope>);
     terminateWorkerMock = vi
       .spyOn(mockWorkerService, 'terminate')
       .mockResolvedValue(undefined);