refactor(ocap-kernel): Use JSON-RPC notifications for vat syscalls (#528)

Fixes #525 

Replaces our janky use of JSON-RPC requests with notifications in the
`VatSupervisor`. Also makes some changes with a view to recognizing
that vat syscalls are handled synchronously by the kernel.

Author: rekmarks

packages/kernel-rpc-methods/src/types.ts

@@ -12,8 +12,8 @@ export type MethodSignature<
 export type MethodSpec<
   Method extends string,
   Params extends JsonRpcParams,
-  Result extends Json | Promise<Json> | void,
-> = Result extends void
+  Result extends Json | Promise<Json> | void | Promise<void>,
+> = Result extends void | Promise<void>
   ? { method: Method; params: Struct<Params>; result?: undefined }
   : {
       method: Method;
@@ -74,7 +74,7 @@ export type ExtractResult<
 
 export type HandlerFunction<
   Params extends JsonRpcParams,
-  Result extends Json | Promise<Json> | void,
+  Result extends Json | Promise<Json> | void | Promise<void>,
   Hooks extends Record<string, unknown>,
 > = (hooks: Hooks, params: Params) => Result;
 
@@ -83,7 +83,7 @@ export type HandlerFunction<
 export type Handler<
   Method extends string,
   Params extends JsonRpcParams,
-  Result extends Json | Promise<Json> | void,
+  Result extends Json | Promise<Json> | void | Promise<void>,
   Hooks extends Record<string, unknown>,
 > = MethodSpec<Method, Params, Result> & {
   hooks: { [Key in keyof Hooks]: true };

packages/logger/src/stream.ts

@@ -3,7 +3,6 @@ import type { JsonRpcCall, JsonRpcMessage } from '@metamask/kernel-utils';
 import { split } from '@metamask/streams';
 import type { DuplexStream } from '@metamask/streams';
 import { isJsonRpcNotification } from '@metamask/utils';
-import type { JsonRpcRequest } from '@metamask/utils';
 
 import type { LogEntry, LogLevel } from './types.ts';
 
@@ -84,7 +83,7 @@ harden(isLoggerMessage);
  */
 export const isKernelMessage = (
   message: JsonRpcMessage,
-): message is JsonRpcRequest => !isLoggerMessage(message);
+): message is JsonRpcMessage => !isLoggerMessage(message);
 harden(isKernelMessage);
 
 /**

packages/ocap-kernel/src/VatHandle.test.ts

@@ -90,7 +90,30 @@ describe('VatHandle', () => {
       await delay(10);
       expect(logger.error).toHaveBeenCalledWith(
         'Unexpected read error',
-        expect.any(Error),
+        expect.objectContaining({
+          message: expect.stringMatching(/Message failed type validation/u),
+        }),
+      );
+    });
+
+    it('throws if handleMessage throws', async () => {
+      const logger = {
+        error: vi.fn(),
+        subLogger: vi.fn(() => logger),
+      } as unknown as Logger;
+      const { stream } = await makeVat({ logger });
+      await stream.receiveInput({
+        id: 'v0:1',
+        method: 'ping',
+        params: [],
+        jsonrpc: '2.0',
+      });
+      await delay(10);
+      expect(logger.error).toHaveBeenCalledWith(
+        'Unexpected read error',
+        expect.objectContaining({
+          message: expect.stringMatching(/^Received unexpected message/u),
+        }),
       );
     });
   });

packages/ocap-kernel/src/VatHandle.ts

@@ -11,9 +11,9 @@ import type {
 import type { VatStore } from '@metamask/kernel-store';
 import type { JsonRpcMessage } from '@metamask/kernel-utils';
 import { Logger } from '@metamask/logger';
-import { serializeError } from '@metamask/rpc-errors';
 import type { DuplexStream } from '@metamask/streams';
-import { isJsonRpcRequest, isJsonRpcResponse } from '@metamask/utils';
+import { isJsonRpcNotification, isJsonRpcResponse } from '@metamask/utils';
+import type { JsonRpcNotification, JsonRpcResponse } from '@metamask/utils';
 
 import type { KernelQueue } from './KernelQueue.ts';
 import { vatMethodSpecs, vatSyscallHandlers } from './rpc/index.ts';
@@ -30,10 +30,14 @@ import type {
 } from './types.ts';
 import { VatSyscall } from './VatSyscall.ts';
 
+type MessageFromVat = JsonRpcResponse | JsonRpcNotification;
+
+type VatStream = DuplexStream<MessageFromVat, JsonRpcMessage>;
+
 type VatConstructorProps = {
   vatId: VatId;
   vatConfig: VatConfig;
-  vatStream: DuplexStream<JsonRpcMessage, JsonRpcMessage>;
+  vatStream: VatStream;
   kernelStore: KernelStore;
   kernelQueue: KernelQueue;
   logger?: Logger | undefined;
@@ -44,7 +48,7 @@ export class VatHandle {
   readonly vatId: VatId;
 
   /** Communications channel to and from the vat itself */
-  readonly #vatStream: DuplexStream<JsonRpcMessage, JsonRpcMessage>;
+  readonly #vatStream: VatStream;
 
   /** The vat's configuration */
   readonly config: VatConfig;
@@ -110,8 +114,8 @@ export class VatHandle {
       `${this.vatId}:`,
     );
     this.#rpcService = new RpcService(vatSyscallHandlers, {
-      handleSyscall: async (params) => {
-        return this.#vatSyscall.handleSyscall(params as VatSyscallObject);
+      handleSyscall: (params) => {
+        this.#vatSyscall.handleSyscall(params as VatSyscallObject);
       },
     });
   }
@@ -181,25 +185,13 @@ export class VatHandle {
   async #handleMessage(message: JsonRpcMessage): Promise<void> {
     if (isJsonRpcResponse(message)) {
       this.#rpcClient.handleResponse(message.id as string, message);
-    } else if (isJsonRpcRequest(message)) {
-      try {
-        this.#rpcService.assertHasMethod(message.method);
-        const result = await this.#rpcService.execute(
-          message.method,
-          message.params,
-        );
-        await this.#vatStream.write({
-          id: message.id,
-          result,
-          jsonrpc: '2.0',
-        });
-      } catch (error) {
-        await this.#vatStream.write({
-          id: message.id,
-          error: serializeError(error),
-          jsonrpc: '2.0',
-        });
-      }
+    } else if (isJsonRpcNotification(message)) {
+      this.#rpcService.assertHasMethod(message.method);
+      await this.#rpcService.execute(message.method, message.params);
+    } else {
+      // We don't expect any JSON-RPC requests from the vat, but the stream may permit them
+      // eslint-disable-next-line @typescript-eslint/restrict-template-expressions
+      throw new Error(`Received unexpected message: ${message}`);
     }
   }
 

packages/ocap-kernel/src/VatSupervisor.ts

@@ -31,6 +31,11 @@ const makeLiveSlots: MakeLiveSlotsFn = localMakeLiveSlots;
 // eslint-disable-next-line n/no-unsupported-features/node-builtins
 export type FetchBlob = (bundleURL: string) => Promise<Response>;
 
+type SupervisorRpcClient = Pick<
+  RpcClient<typeof vatSyscallMethodSpecs>,
+  'notify' | 'handleResponse'
+>;
+
 type SupervisorConstructorProps = {
   id: VatId;
   kernelStream: DuplexStream<JsonRpcMessage, JsonRpcMessage>;
@@ -54,7 +59,7 @@ export class VatSupervisor {
   readonly #logger: Logger;
 
   /** RPC client for sending syscall requests to the kernel */
-  readonly #rpcClient: RpcClient<typeof vatSyscallMethodSpecs>;
+  readonly #rpcClient: SupervisorRpcClient;
 
   /** RPC service for handling requests from the kernel */
   readonly #rpcService: RpcService<typeof vatHandlers>;
@@ -74,9 +79,6 @@ export class VatSupervisor {
   /** Capability to fetch the bundle of code to run in this vat. */
   readonly #fetchBlob: FetchBlob;
 
-  /** Result promises from all syscalls sent to the kernel in the current crank */
-  readonly #syscallsInFlight: Promise<unknown>[] = [];
-
   /**
    * Construct a new VatSupervisor instance.
    *
@@ -177,17 +179,15 @@ export class VatSupervisor {
    * @returns a syscall success result.
    */
   executeSyscall(vso: VatSyscallObject): VatSyscallResult {
-    this.#syscallsInFlight.push(
-      // IMPORTANT: Syscall architecture design explanation:
-      // - Vats operate on an "optimistic execution" model - they send syscalls and continue execution
-      //    without waiting for responses, assuming success.
-      // - The Kernel processes syscalls asynchronously and failures are catched in VatHandle.
-      this.#rpcClient
-        .call('syscall', coerceVatSyscallObject(vso))
-        // We catch these rejections here to prevent unhandled promise rejections,
-        // as they're an expected part of the architecture, not errors
-        .catch(() => undefined),
-    );
+    // IMPORTANT: Syscall architecture design explanation:
+    // - Vats operate on an "optimistic execution" model - they send syscalls and continue execution
+    //    without waiting for responses, assuming success.
+    // - The Kernel processes syscalls synchronously on receipt and failures are caught in VatHandle.
+    // - The vat is terminated and the crank is rolled back if a syscall fails.
+    this.#rpcClient
+      .notify('syscall', coerceVatSyscallObject(vso))
+      // Just to please the linter (notifications never reject)
+      .catch(() => undefined);
     return ['ok', null];
   }
 
@@ -204,13 +204,6 @@ export class VatSupervisor {
       // Handle delivery errors
       deliveryError = error instanceof Error ? error.message : String(error);
       this.#logger.error(`Delivery error in vat ${this.id}:`, deliveryError);
-    } finally {
-      // Clean up at the end of a crank
-      this.#syscallsInFlight.length = 0;
-      // Reject all pending RPC requests to maintain the optimistic execution model
-      // This prevents late responses from affecting the vat in unexpected ways
-      // between cranks.
-      this.#rpcClient.rejectAll(new Error('end of crank'));
     }
 
     // eslint-disable-next-line @typescript-eslint/no-non-null-assertion

packages/ocap-kernel/src/rpc/vat-syscall/vat-syscall.test.ts

@@ -1,33 +1,35 @@
 import { describe, it, expect, vi } from 'vitest';
 
 import { vatSyscallHandler } from './vat-syscall.ts';
-import type { HandleSyscall } from './vat-syscall.ts';
 
 describe('vatSyscallHandler', () => {
-  it('should initialize a vat', async () => {
-    const handleSyscall = vi.fn(async () =>
-      Promise.resolve({
-        checkpoint: 'fake',
-      }),
-    ) as unknown as HandleSyscall;
-    const result = await vatSyscallHandler.implementation({ handleSyscall }, [
+  it('should initialize a vat', () => {
+    const handleSyscall = vi.fn();
+    vatSyscallHandler.implementation({ handleSyscall }, [
+      'send',
+      'test',
+      {
+        methargs: { body: 'test', slots: [] },
+        result: null,
+      },
+    ]);
+
+    expect(handleSyscall).toHaveBeenCalledTimes(1);
+    expect(handleSyscall).toHaveBeenCalledWith([
       'send',
       'test',
       {
         methargs: { body: 'test', slots: [] },
         result: null,
       },
     ]);
-    expect(result).toStrictEqual({
-      checkpoint: 'fake',
-    });
   });
 
-  it('should propagate errors from hooks', async () => {
+  it('should propagate errors from hooks', () => {
     const handleSyscall = vi.fn(() => {
       throw new Error('fake');
     });
-    await expect(
+    expect(() =>
       vatSyscallHandler.implementation({ handleSyscall }, [
         'send',
         'test',
@@ -36,6 +38,6 @@ describe('vatSyscallHandler', () => {
           result: null,
         },
       ]),
-    ).rejects.toThrow('fake');
+    ).toThrow('fake');
   });
 });

packages/ocap-kernel/src/rpc/vat-syscall/vat-syscall.ts

@@ -1,4 +1,3 @@
-import type { VatSyscallResult } from '@agoric/swingset-liveslots';
 import type { Handler, MethodSpec } from '@metamask/kernel-rpc-methods';
 import {
   tuple,
@@ -7,7 +6,6 @@ import {
   string,
   union,
   boolean,
-  Struct,
 } from '@metamask/superstruct';
 import type { Infer } from '@metamask/superstruct';
 
@@ -65,27 +63,12 @@ const VatSyscallParamsStruct = union([
 
 type VatSyscallParams = Infer<typeof VatSyscallParamsStruct>;
 
-const VatSyscallResultStruct: Struct<VatSyscallResult> = union([
-  tuple([
-    literal('ok'),
-    union([CapDataStruct, string(), array(string()), literal(null)]),
-  ]),
-  tuple([literal('error'), string()]),
-]);
-
-export const vatSyscallSpec: MethodSpec<
-  'syscall',
-  VatSyscallParams,
-  Promise<VatSyscallResult>
-> = {
+export const vatSyscallSpec: MethodSpec<'syscall', VatSyscallParams, void> = {
   method: 'syscall',
   params: VatSyscallParamsStruct,
-  result: VatSyscallResultStruct,
 } as const;
 
-export type HandleSyscall = (
-  params: VatSyscallParams,
-) => Promise<VatSyscallResult>;
+export type HandleSyscall = (params: VatSyscallParams) => void;
 
 type SyscallHooks = {
   handleSyscall: HandleSyscall;
@@ -94,12 +77,12 @@ type SyscallHooks = {
 export const vatSyscallHandler: Handler<
   'syscall',
   VatSyscallParams,
-  Promise<VatSyscallResult>,
+  void,
   SyscallHooks
 > = {
   ...vatSyscallSpec,
   hooks: { handleSyscall: true },
-  implementation: async ({ handleSyscall }, params) => {
-    return await handleSyscall(params);
+  implementation: ({ handleSyscall }, params) => {
+    handleSyscall(params);
   },
 } as const;

vitest.config.ts

@@ -134,10 +134,10 @@ export default defineConfig({
           lines: 73.58,
         },
         'packages/ocap-kernel/**': {
-          statements: 91.36,
+          statements: 91.39,
           functions: 95.07,
-          branches: 81.99,
-          lines: 91.33,
+          branches: 82.16,
+          lines: 91.37,
         },
         'packages/streams/**': {
           statements: 100,