refactor: Support JSON-RPC notifications in addition to requests (#496)

Closes #368 

Adds support for JSON-RPC notifications in addition to JSON-RPC
requests. Adds a new method, `RpcClient.notify()`, for this purpose.
Updates stream types to permit notifications in additions to request. We
do not make use of this capability for any method right now, but may do
so for #488.

Author: rekmarks

packages/extension/src/kernel-integration/VatWorkerClient.ts

@@ -1,9 +1,5 @@
 import { isJsonRpcResponse } from '@metamask/utils';
-import type {
-  JsonRpcId,
-  JsonRpcRequest,
-  JsonRpcResponse,
-} from '@metamask/utils';
+import type { JsonRpcId, JsonRpcResponse } from '@metamask/utils';
 import type { VatWorkerManager, VatId, VatConfig } from '@ocap/kernel';
 import { vatWorkerServiceMethodSpecs } from '@ocap/kernel/rpc';
 import { RpcClient } from '@ocap/rpc-methods';
@@ -16,7 +12,7 @@ import type {
   PostMessageEnvelope,
   PostMessageTarget,
 } from '@ocap/streams/browser';
-import type { JsonRpcMessage, Logger } from '@ocap/utils';
+import type { JsonRpcCall, JsonRpcMessage, Logger } from '@ocap/utils';
 import { isJsonRpcMessage, makeLogger, stringify } from '@ocap/utils';
 
 // Appears in the docs.
@@ -25,7 +21,7 @@ import type { ExtensionVatWorkerService } from './VatWorkerServer.ts';
 
 export type VatWorkerClientStream = PostMessageDuplexStream<
   MessageEvent<JsonRpcResponse>,
-  PostMessageEnvelope<JsonRpcRequest>
+  PostMessageEnvelope<JsonRpcCall>
 >;
 
 export class ExtensionVatWorkerClient implements VatWorkerManager {
@@ -61,8 +57,10 @@ export class ExtensionVatWorkerClient implements VatWorkerManager {
     this.#rpcClient = new RpcClient(
       vatWorkerServiceMethodSpecs,
       async (request) => {
-        if (request.method === 'launch') {
-          this.#portMap.set(request.id, undefined);
+        if ('id' in request) {
+          if (request.method === 'launch') {
+            this.#portMap.set(request.id, undefined);
+          }
         }
         await this.#stream.write({ payload: request, transfer: [] });
       },

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

@@ -1,6 +1,5 @@
 import { JsonRpcEngine } from '@metamask/json-rpc-engine';
 import type { JsonRpcRequest, JsonRpcResponse } from '@metamask/utils';
-import { isJsonRpcRequest } from '@metamask/utils';
 import type { ClusterConfig } from '@ocap/kernel';
 import { ClusterConfigStruct, Kernel } from '@ocap/kernel';
 import { makeSQLKernelDatabase } from '@ocap/store/sqlite/wasm';
@@ -9,7 +8,8 @@ import {
   MessagePortDuplexStream,
   receiveMessagePort,
 } from '@ocap/streams/browser';
-import { fetchValidatedJson, Logger } from '@ocap/utils';
+import { fetchValidatedJson, isJsonRpcCall, Logger } from '@ocap/utils';
+import type { JsonRpcCall } from '@ocap/utils';
 
 import { makeLoggingMiddleware } from './middleware/logging.ts';
 import { createPanelMessageMiddleware } from './middleware/panel-message.ts';
@@ -39,9 +39,9 @@ async function main(): Promise<void> {
   );
 
   const kernelStream = await MessagePortDuplexStream.make<
-    JsonRpcRequest,
+    JsonRpcCall,
     JsonRpcResponse
-  >(port, isJsonRpcRequest);
+  >(port, isJsonRpcCall);
 
   // Initialize kernel dependencies
   const vatWorkerClient = ExtensionVatWorkerClient.make(
@@ -63,7 +63,11 @@ async function main(): Promise<void> {
   const kernelEngine = new JsonRpcEngine();
   kernelEngine.push(makeLoggingMiddleware(logger.subLogger('kernel-command')));
   kernelEngine.push(createPanelMessageMiddleware(kernel, kernelDatabase));
-  receiveUiConnections(async (request) => kernelEngine.handle(request), logger);
+  // JsonRpcEngine type error: does not handle JSON-RPC notifications
+  receiveUiConnections(
+    async (request) => kernelEngine.handle(request as JsonRpcRequest),
+    logger,
+  );
   const launchDefaultSubcluster = firstTime || ALWAYS_RESET_STORAGE;
 
   const defaultSubcluster = await fetchValidatedJson<ClusterConfig>(

packages/extension/src/kernel-integration/ui-connections.test.ts

@@ -1,8 +1,8 @@
-import type { JsonRpcRequest, JsonRpcResponse } from '@metamask/utils';
+import type { JsonRpcResponse } from '@metamask/utils';
 import type { PostMessageTarget } from '@ocap/streams/browser';
 import { delay } from '@ocap/test-utils';
 import { TestDuplexStream } from '@ocap/test-utils/streams';
-import type { Logger } from '@ocap/utils';
+import type { JsonRpcCall, Logger } from '@ocap/utils';
 import { describe, it, expect, vi, beforeEach } from 'vitest';
 
 import {
@@ -161,7 +161,7 @@ describe('ui-connections', () => {
     const logger = makeMockLogger();
 
     const mockHandleMessage = vi.fn(
-      async (_request: JsonRpcRequest): Promise<JsonRpcResponse> => ({
+      async (_request: JsonRpcCall): Promise<JsonRpcResponse> => ({
         id: 'foo',
         jsonrpc: '2.0' as const,
         result: { vats: [], clusterConfig },

packages/extension/src/kernel-integration/ui-connections.ts

@@ -1,8 +1,8 @@
 import { isJsonRpcRequest, isJsonRpcResponse } from '@metamask/utils';
-import type { JsonRpcRequest, JsonRpcResponse } from '@metamask/utils';
+import type { JsonRpcResponse } from '@metamask/utils';
 import { PostMessageDuplexStream } from '@ocap/streams/browser';
 import { stringify } from '@ocap/utils';
-import type { Logger } from '@ocap/utils';
+import type { JsonRpcCall, Logger } from '@ocap/utils';
 import { nanoid } from 'nanoid';
 
 import { isUiControlCommand } from './ui-control-command.ts';
@@ -11,18 +11,16 @@ import type { UiControlCommand } from './ui-control-command.ts';
 export const UI_CONTROL_CHANNEL_NAME = 'ui-control';
 
 export type KernelControlStream = PostMessageDuplexStream<
-  JsonRpcRequest,
+  JsonRpcCall,
   JsonRpcResponse
 >;
 
 export type KernelControlReplyStream = PostMessageDuplexStream<
   JsonRpcResponse,
-  JsonRpcRequest
+  JsonRpcCall
 >;
 
-type HandleInstanceMessage = (
-  request: JsonRpcRequest,
-) => Promise<JsonRpcResponse>;
+type HandleInstanceMessage = (request: JsonRpcCall) => Promise<JsonRpcResponse>;
 
 /**
  * Establishes a connection between a UI instance and the kernel. Should be called
@@ -45,7 +43,7 @@ export const establishKernelConnection = async (
 
   const kernelStream = await PostMessageDuplexStream.make<
     JsonRpcResponse,
-    JsonRpcRequest
+    JsonRpcCall
   >({
     validateInput: isJsonRpcResponse,
     messageTarget: instanceChannel,

packages/extension/src/offscreen.ts

@@ -1,13 +1,14 @@
-import { isJsonRpcRequest, isJsonRpcResponse } from '@metamask/utils';
-import type { JsonRpcRequest, JsonRpcResponse } from '@metamask/utils';
+import { isJsonRpcResponse } from '@metamask/utils';
+import type { JsonRpcResponse } from '@metamask/utils';
 import type { DuplexStream } from '@ocap/streams';
 import {
   initializeMessageChannel,
   ChromeRuntimeDuplexStream,
   MessagePortDuplexStream,
 } from '@ocap/streams/browser';
 import type { PostMessageTarget } from '@ocap/streams/browser';
-import { delay, Logger } from '@ocap/utils';
+import { delay, isJsonRpcCall, Logger } from '@ocap/utils';
+import type { JsonRpcCall } from '@ocap/utils';
 
 import { makeIframeVatWorker } from './kernel-integration/iframe-vat-worker.ts';
 import { ExtensionVatWorkerService } from './kernel-integration/VatWorkerServer.ts';
@@ -25,9 +26,9 @@ async function main(): Promise<void> {
 
   // Create stream for messages from the background script
   const backgroundStream = await ChromeRuntimeDuplexStream.make<
-    JsonRpcRequest,
+    JsonRpcCall,
     JsonRpcResponse
-  >(chrome.runtime, 'offscreen', 'background', isJsonRpcRequest);
+  >(chrome.runtime, 'offscreen', 'background', isJsonRpcCall);
 
   const { kernelStream, vatWorkerService } = await makeKernelWorker();
 
@@ -45,7 +46,7 @@ async function main(): Promise<void> {
  * @returns The message port stream for worker communication
  */
 async function makeKernelWorker(): Promise<{
-  kernelStream: DuplexStream<JsonRpcResponse, JsonRpcRequest>;
+  kernelStream: DuplexStream<JsonRpcResponse, JsonRpcCall>;
   vatWorkerService: ExtensionVatWorkerService;
 }> {
   const worker = new Worker('kernel-worker.js', { type: 'module' });
@@ -56,7 +57,7 @@ async function makeKernelWorker(): Promise<{
 
   const kernelStream = await MessagePortDuplexStream.make<
     JsonRpcResponse,
-    JsonRpcRequest
+    JsonRpcCall
   >(port, isJsonRpcResponse);
 
   const vatWorkerService = ExtensionVatWorkerService.make(

packages/kernel/src/Kernel.ts

@@ -1,6 +1,7 @@
 import type { CapData } from '@endo/marshal';
 import { serializeError } from '@metamask/rpc-errors';
-import type { JsonRpcRequest, JsonRpcResponse } from '@metamask/utils';
+import { hasProperty } from '@metamask/utils';
+import type { JsonRpcResponse } from '@metamask/utils';
 import {
   StreamReadError,
   VatAlreadyExistsError,
@@ -11,6 +12,7 @@ import type { ExtractParams, ExtractResult } from '@ocap/rpc-methods';
 import type { KernelDatabase } from '@ocap/store';
 import type { DuplexStream } from '@ocap/streams';
 import { Logger } from '@ocap/utils';
+import type { JsonRpcCall } from '@ocap/utils';
 
 import { KernelQueue } from './KernelQueue.ts';
 import { KernelRouter } from './KernelRouter.ts';
@@ -33,7 +35,7 @@ import { VatHandle } from './VatHandle.ts';
 
 export class Kernel {
   /** Command channel from the controlling console/browser extension/test driver */
-  readonly #commandStream: DuplexStream<JsonRpcRequest, JsonRpcResponse>;
+  readonly #commandStream: DuplexStream<JsonRpcCall, JsonRpcResponse>;
 
   readonly #rpcService: RpcService<typeof kernelHandlers>;
 
@@ -70,7 +72,7 @@ export class Kernel {
    */
   // eslint-disable-next-line no-restricted-syntax
   private constructor(
-    commandStream: DuplexStream<JsonRpcRequest, JsonRpcResponse>,
+    commandStream: DuplexStream<JsonRpcCall, JsonRpcResponse>,
     vatWorkerService: VatWorkerManager,
     kernelDatabase: KernelDatabase,
     options: {
@@ -108,7 +110,7 @@ export class Kernel {
    * @returns A promise for the new kernel instance.
    */
   static async make(
-    commandStream: DuplexStream<JsonRpcRequest, JsonRpcResponse>,
+    commandStream: DuplexStream<JsonRpcCall, JsonRpcResponse>,
     vatWorkerService: VatWorkerManager,
     kernelDatabase: KernelDatabase,
     options: {
@@ -155,25 +157,29 @@ export class Kernel {
    *
    * @param message - The message to handle.
    */
-  async #handleCommandMessage(message: JsonRpcRequest): Promise<void> {
+  async #handleCommandMessage(message: JsonRpcCall): Promise<void> {
     try {
       this.#rpcService.assertHasMethod(message.method);
       const result = await this.#rpcService.execute(
         message.method,
         message.params,
       );
-      await this.#commandStream.write({
-        id: message.id,
-        jsonrpc: '2.0',
-        result,
-      });
+      if (hasProperty(message, 'id') && typeof message.id === 'string') {
+        await this.#commandStream.write({
+          id: message.id,
+          jsonrpc: '2.0',
+          result,
+        });
+      }
     } catch (error) {
       this.#logger.error('Error executing command', error);
-      await this.#commandStream.write({
-        id: message.id,
-        jsonrpc: '2.0',
-        error: serializeError(error),
-      });
+      if (hasProperty(message, 'id') && typeof message.id === 'string') {
+        await this.#commandStream.write({
+          id: message.id,
+          jsonrpc: '2.0',
+          error: serializeError(error),
+        });
+      }
     }
   }
 

packages/kernel/src/rpc/kernel/index.ts

@@ -1,14 +1,18 @@
-import type { MethodRequest } from '@ocap/rpc-methods';
+import type {
+  HandlerRecord,
+  MethodRequest,
+  MethodSpecRecord,
+} from '@ocap/rpc-methods';
 
 import { pingHandler, pingSpec } from '../vat/ping.ts';
 
 export const kernelHandlers = {
   ping: pingHandler,
-} as const;
+} as HandlerRecord<typeof pingHandler>;
 
 export const kernelMethodSpecs = {
   ping: pingSpec,
-} as const;
+} as MethodSpecRecord<typeof pingSpec>;
 
 type Handlers = (typeof kernelHandlers)[keyof typeof kernelHandlers];
 

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

@@ -1,12 +1,14 @@
+import type { MethodSpecRecord, HandlerRecord } from '@ocap/rpc-methods';
+
 import { vatSyscallSpec, vatSyscallHandler } from './vat-syscall.ts';
 
 export const vatSyscallHandlers = {
   syscall: vatSyscallHandler,
-} as const;
+} as HandlerRecord<typeof vatSyscallHandler>;
 
 export const vatSyscallMethodSpecs = {
   syscall: vatSyscallSpec,
-} as const;
+} as MethodSpecRecord<typeof vatSyscallSpec>;
 
 type Handlers = (typeof vatSyscallHandlers)[keyof typeof vatSyscallHandlers];
 

packages/rpc-methods/src/RpcClient.test.ts

@@ -69,6 +69,33 @@ describe('RpcClient', () => {
     });
   });
 
+  describe('notify', () => {
+    it('should call a notification method', async () => {
+      const sendMessage = vi.fn(async () => Promise.resolve());
+      const client = new RpcClient(getMethods(), sendMessage, 'test');
+      await client.notify('method3', ['test']);
+      expect(sendMessage).toHaveBeenCalledWith({
+        jsonrpc: jsonrpc2,
+        method: 'method3',
+        params: ['test'],
+      });
+    });
+
+    it('should log an error if the message fails to send', async () => {
+      const logger = makeLogger('[test]');
+      const sendMessage = vi.fn(async () =>
+        Promise.reject(new Error('test error')),
+      );
+      const client = new RpcClient(getMethods(), sendMessage, 'test', logger);
+      const logError = vi.spyOn(logger, 'error');
+      await client.notify('method3', ['test']);
+      expect(logError).toHaveBeenCalledWith(
+        'Failed to send notification',
+        new Error('test error'),
+      );
+    });
+  });
+
   describe('callAndGetId', () => {
     it('should call a method and return the id', async () => {
       const client = new RpcClient(getMethods(), vi.fn(), 'test');