feat: Add timeout handling for remote message sends and URL redemptions

Author: sirtimid

packages/ocap-kernel/src/remotes/RemoteHandle.test.ts

@@ -1,6 +1,7 @@
 import type { VatOneResolution } from '@agoric/swingset-liveslots';
 import type { Logger } from '@metamask/logger';
-import { describe, it, expect, vi, beforeEach } from 'vitest';
+import { makeAbortSignalMock } from '@ocap/repo-tools/test-utils';
+import { describe, it, expect, vi, beforeEach, afterEach } from 'vitest';
 
 import type { KernelQueue } from '../KernelQueue.ts';
 import { RemoteHandle } from './RemoteHandle.ts';
@@ -626,4 +627,119 @@ describe('RemoteHandle', () => {
     // Verify they resolved independently (different values)
     expect(kref1).not.toBe(kref2);
   });
+
+  describe('redeemOcapURL timeout', () => {
+    afterEach(() => {
+      vi.restoreAllMocks();
+    });
+
+    it('sets up 30-second timeout using AbortSignal.timeout', async () => {
+      const remote = makeRemote();
+      const mockOcapURL = 'ocap:test@peer';
+
+      let mockSignal: ReturnType<typeof makeAbortSignalMock> | undefined;
+      vi.spyOn(AbortSignal, 'timeout').mockImplementation((ms: number) => {
+        mockSignal = makeAbortSignalMock(ms);
+        return mockSignal;
+      });
+
+      const urlPromise = remote.redeemOcapURL(mockOcapURL);
+
+      // Verify AbortSignal.timeout was called with 30 seconds
+      expect(AbortSignal.timeout).toHaveBeenCalledWith(30_000);
+      expect(mockSignal?.timeoutMs).toBe(30_000);
+
+      // Resolve the redemption to avoid hanging
+      const sendCall = vi.mocked(mockRemoteComms.sendRemoteMessage).mock
+        .calls[0];
+      const sentMessage = JSON.parse(sendCall?.[1] as string);
+      const replyKey = sentMessage.params[1] as string;
+
+      await remote.handleRemoteMessage(
+        JSON.stringify({
+          method: 'redeemURLReply',
+          params: [true, replyKey, 'ro+1'],
+        }),
+      );
+
+      await urlPromise;
+    });
+
+    it('cleans up pending redemption when redemption succeeds before timeout', async () => {
+      const remote = makeRemote();
+      const mockOcapURL = 'ocap:test@peer';
+      const mockURLResolutionRRef = 'ro+6';
+      const mockURLResolutionKRef = 'ko1';
+      const expectedReplyKey = '1';
+
+      let mockSignal: ReturnType<typeof makeAbortSignalMock> | undefined;
+      vi.spyOn(AbortSignal, 'timeout').mockImplementation((ms: number) => {
+        mockSignal = makeAbortSignalMock(ms);
+        return mockSignal;
+      });
+
+      const urlPromise = remote.redeemOcapURL(mockOcapURL);
+
+      // Send reply immediately (before timeout)
+      const redeemURLReply = {
+        method: 'redeemURLReply',
+        params: [true, expectedReplyKey, mockURLResolutionRRef],
+      };
+      await remote.handleRemoteMessage(JSON.stringify(redeemURLReply));
+
+      const kref = await urlPromise;
+      expect(kref).toBe(mockURLResolutionKRef);
+
+      // Verify timeout signal was not aborted
+      expect(mockSignal?.aborted).toBe(false);
+
+      // Verify cleanup happened - trying to handle another reply with the same key should fail
+      await expect(
+        remote.handleRemoteMessage(JSON.stringify(redeemURLReply)),
+      ).rejects.toThrow(`unknown URL redemption reply key ${expectedReplyKey}`);
+    });
+
+    it('cleans up pending redemption map entry on timeout', async () => {
+      const remote = makeRemote();
+      const mockOcapURL = 'ocap:test@peer';
+
+      let mockSignal: ReturnType<typeof makeAbortSignalMock> | undefined;
+      vi.spyOn(AbortSignal, 'timeout').mockImplementation((ms: number) => {
+        mockSignal = makeAbortSignalMock(ms);
+        return mockSignal;
+      });
+
+      // Start a redemption
+      const urlPromise = remote.redeemOcapURL(mockOcapURL);
+
+      // Get the reply key that was used
+      const sendCall = vi.mocked(mockRemoteComms.sendRemoteMessage).mock
+        .calls[0];
+      const sentMessage = JSON.parse(sendCall?.[1] as string);
+      const replyKey = sentMessage.params[1] as string;
+
+      // Wait for the promise to be set up and event listener registered
+      await new Promise<void>((resolve) => queueMicrotask(() => resolve()));
+
+      // Manually trigger the abort to simulate timeout
+      mockSignal?.abort();
+
+      // Wait for the abort handler to execute
+      await new Promise<void>((resolve) => queueMicrotask(() => resolve()));
+
+      // Verify the promise rejects
+      await expect(urlPromise).rejects.toThrow(
+        'URL redemption timed out after 30 seconds',
+      );
+
+      // Verify cleanup happened - trying to handle a reply with the same key should fail
+      const redeemURLReply = {
+        method: 'redeemURLReply',
+        params: [true, replyKey, 'ro+1'],
+      };
+      await expect(
+        remote.handleRemoteMessage(JSON.stringify(redeemURLReply)),
+      ).rejects.toThrow(`unknown URL redemption reply key ${replyKey}`);
+    });
+  });
 });

packages/ocap-kernel/src/remotes/RemoteHandle.ts

@@ -460,13 +460,34 @@ export class RemoteHandle implements EndpointHandle {
     const replyKey = `${this.#redemptionCounter}`;
     this.#redemptionCounter += 1;
     const { promise, resolve, reject } = makePromiseKit<string>();
-    // XXX TODO: Probably these should have timeouts
     this.#pendingRedemptions.set(replyKey, [resolve, reject]);
-    await this.#sendRemoteCommand({
-      method: 'redeemURL',
-      params: [url, replyKey],
+
+    // Set up timeout handling with AbortSignal
+    const timeoutSignal = AbortSignal.timeout(30_000);
+    const timeoutPromise = new Promise<never>((_resolve, _reject) => {
+      timeoutSignal.addEventListener('abort', () => {
+        // Clean up from pending redemptions map
+        if (this.#pendingRedemptions.has(replyKey)) {
+          this.#pendingRedemptions.delete(replyKey);
+        }
+        _reject(new Error('URL redemption timed out after 30 seconds'));
+      });
     });
-    return promise;
+
+    try {
+      await this.#sendRemoteCommand({
+        method: 'redeemURL',
+        params: [url, replyKey],
+      });
+      // Wait for reply with timeout protection
+      return await Promise.race([promise, timeoutPromise]);
+    } catch (error) {
+      // Clean up and remove from map if still pending
+      if (this.#pendingRedemptions.has(replyKey)) {
+        this.#pendingRedemptions.delete(replyKey);
+      }
+      throw error;
+    }
   }
 
   /**

packages/ocap-kernel/src/remotes/network.test.ts

@@ -1,4 +1,5 @@
 import { AbortError } from '@metamask/kernel-errors';
+import { makeAbortSignalMock } from '@ocap/repo-tools/test-utils';
 import {
   describe,
   expect,

packages/ocap-kernel/src/remotes/network.ts

@@ -100,6 +100,31 @@ export async function initNetwork(
     return queue;
   }
 
+  /**
+   * Write a message to a channel stream with a timeout.
+   *
+   * @param channel - The channel to write to.
+   * @param message - The message bytes to write.
+   * @param timeoutMs - Timeout in milliseconds (default: 10 seconds).
+   * @returns Promise that resolves when the write completes or rejects on timeout.
+   * @throws Error if the write times out or fails.
+   */
+  async function writeWithTimeout(
+    channel: Channel,
+    message: Uint8Array,
+    timeoutMs = 10_000,
+  ): Promise<void> {
+    const timeoutSignal = AbortSignal.timeout(timeoutMs);
+    return Promise.race([
+      channel.msgStream.write(message),
+      new Promise<never>((_resolve, reject) => {
+        timeoutSignal.addEventListener('abort', () => {
+          reject(new Error(`Message send timed out after ${timeoutMs}ms`));
+        });
+      }),
+    ]);
+  }
+
   /**
    * Receive a message from a peer.
    *
@@ -315,7 +340,7 @@ export async function initNetwork(
     while ((queuedMsg = queue.dequeue()) !== undefined) {
       try {
         logger.log(`${peerId}:: send (queued) ${queuedMsg.message}`);
-        await channel.msgStream.write(fromString(queuedMsg.message));
+        await writeWithTimeout(channel, fromString(queuedMsg.message), 10_000);
       } catch (problem) {
         outputError(peerId, `sending queued message`, problem);
         // Preserve the failed message and all remaining messages
@@ -397,7 +422,7 @@ export async function initNetwork(
 
     try {
       logger.log(`${targetPeerId}:: send ${message}`);
-      await channel.msgStream.write(fromString(message));
+      await writeWithTimeout(channel, fromString(message), 10_000);
       reconnectionManager.resetBackoff(targetPeerId);
     } catch (problem) {
       outputError(targetPeerId, `sending message`, problem);

packages/repo-tools/src/test-utils/abort-signal.ts

@@ -0,0 +1,51 @@
+import { vi } from 'vitest';
+
+/**
+ * Create a mock AbortSignal that can be manually aborted.
+ *
+ * @param timeoutMs - The timeout value (stored for verification).
+ * @returns A mock AbortSignal.
+ */
+export function makeAbortSignalMock(timeoutMs: number): AbortSignal & {
+  abort: () => void;
+  timeoutMs: number;
+} {
+  const handlers: (() => void)[] = [];
+  let aborted = false;
+
+  const signal = {
+    get aborted() {
+      return aborted;
+    },
+    timeoutMs,
+    addEventListener: vi.fn((event: string, handler: () => void) => {
+      if (event === 'abort') {
+        handlers.push(handler);
+      }
+    }),
+    removeEventListener: vi.fn((event: string, handler: () => void) => {
+      if (event === 'abort') {
+        const index = handlers.indexOf(handler);
+        if (index > -1) {
+          handlers.splice(index, 1);
+        }
+      }
+    }),
+    dispatchEvent: vi.fn(),
+    onabort: null,
+    reason: undefined,
+    throwIfAborted: vi.fn(),
+    abort() {
+      aborted = true;
+      // Call all handlers synchronously
+      for (const handler of handlers) {
+        handler();
+      }
+    },
+  } as AbortSignal & {
+    abort: () => void;
+    timeoutMs: number;
+  };
+
+  return signal;
+}

packages/repo-tools/src/test-utils/index.ts

@@ -4,6 +4,7 @@ export { makePromiseKitMock } from './promise-kit.ts';
 export { fetchMock } from './env/fetch-mock.ts';
 export * from './env/mock-kernel.ts';
 export { makeMockMessageTarget } from './postMessage.ts';
+export { makeAbortSignalMock } from './abort-signal.ts';
 
 // eslint-disable-next-line @typescript-eslint/ban-ts-comment
 // @ts-ignore Intermittent browser/Node incompatibility

vitest.config.ts

@@ -106,7 +106,7 @@ export default defineConfig({
         'packages/kernel-platforms/**': {
           statements: 99.38,
           functions: 100,
-          branches: 96.2,
+          branches: 96.25,
           lines: 99.38,
         },
         'packages/kernel-rpc-methods/**': {