From a4ede042843f6dcc0fc686adda3d25c0d338cef2 Mon Sep 17 00:00:00 2001
From: Erik Marks <rekmarks@protonmail.com>
Date: Fri, 25 Oct 2024 10:06:41 -0700
Subject: [PATCH 1/4] feat(streams)!: Add optional input validator arg to
 BaseReader

Breaking because we had to convert to an options bag for BaseReader.
---
 packages/streams/src/BaseStream.test.ts     | 13 +++++++++++-
 packages/streams/src/BaseStream.ts          | 19 ++++++++++++++++--
 packages/streams/src/ChromeRuntimeStream.ts |  8 +++++---
 packages/streams/src/MessagePortStream.ts   | 10 ++++++----
 packages/streams/src/PostMessageStream.ts   |  8 +++++---
 packages/streams/test/stream-mocks.ts       | 22 +++++++++++++++------
 6 files changed, 61 insertions(+), 19 deletions(-)

diff --git a/packages/streams/src/BaseStream.test.ts b/packages/streams/src/BaseStream.test.ts
index 756f06d9c..51f60408b 100644
--- a/packages/streams/src/BaseStream.test.ts
+++ b/packages/streams/src/BaseStream.test.ts
@@ -30,7 +30,7 @@ describe('BaseReader', () => {
 
     it('calls onEnd once when ending', async () => {
       const onEnd = vi.fn();
-      const reader = new TestReader(onEnd);
+      const reader = new TestReader({ onEnd });
       expect(onEnd).not.toHaveBeenCalled();
 
       await reader.return();
@@ -50,6 +50,17 @@ describe('BaseReader', () => {
       expect(await reader.next()).toStrictEqual(makePendingResult(message));
     });
 
+    it('calls inputValidator with received input if specified', async () => {
+      const inputValidator = vi.fn();
+      const reader = new TestReader({ inputValidator });
+
+      const message = 42;
+      reader.receiveInput(message);
+
+      expect(await reader.next()).toStrictEqual(makePendingResult(message));
+      expect(inputValidator).toHaveBeenCalledWith(message);
+    });
+
     it('emits message received after next()', async () => {
       const reader = new TestReader();
 
diff --git a/packages/streams/src/BaseStream.ts b/packages/streams/src/BaseStream.ts
index 6cc2f76f0..ee68e46b5 100644
--- a/packages/streams/src/BaseStream.ts
+++ b/packages/streams/src/BaseStream.ts
@@ -94,6 +94,15 @@ harden(makeStreamBuffer);
  */
 export type OnEnd = () => void | Promise<void>;
 
+export type InputValidator<Read extends Json> = (
+  input: Json,
+) => asserts input is Read;
+
+export type BaseReaderArgs<Read extends Json> = {
+  inputValidator?: InputValidator<Read> | undefined;
+  onEnd?: OnEnd | undefined;
+};
+
 /**
  * A function that receives input from a transport mechanism to a readable stream.
  * Validates that the input is an {@link IteratorResult}, and throws if it is not.
@@ -117,6 +126,8 @@ export class BaseReader<Read extends Json> implements Reader<Read> {
    */
   readonly #buffer = makeStreamBuffer<IteratorResult<Read, undefined>>();
 
+  readonly #inputValidator?: InputValidator<Read> | undefined;
+
   #onEnd?: OnEnd | undefined;
 
   #didExposeReceiveInput: boolean = false;
@@ -124,10 +135,13 @@ export class BaseReader<Read extends Json> implements Reader<Read> {
   /**
    * Constructs a {@link BaseReader}.
    *
-   * @param onEnd - A function that is called when the stream ends. For any cleanup that
+   * @param args - Options bag.
+   * @param args.inputValidator - A function that validates input from the transport.
+   * @param args.onEnd - A function that is called when the stream ends. For any cleanup that
    * should happen when the stream ends, such as closing a message port.
    */
-  constructor(onEnd?: () => void) {
+  constructor({ inputValidator, onEnd }: BaseReaderArgs<Read>) {
+    this.#inputValidator = inputValidator;
     this.#onEnd = onEnd;
     harden(this);
   }
@@ -172,6 +186,7 @@ export class BaseReader<Read extends Json> implements Reader<Read> {
       return;
     }
 
+    this.#inputValidator?.(unmarshaled);
     this.#buffer.put(makePendingResult(unmarshaled as Read));
   };
 
diff --git a/packages/streams/src/ChromeRuntimeStream.ts b/packages/streams/src/ChromeRuntimeStream.ts
index 9ad5e8f65..0575feeb8 100644
--- a/packages/streams/src/ChromeRuntimeStream.ts
+++ b/packages/streams/src/ChromeRuntimeStream.ts
@@ -75,9 +75,11 @@ export class ChromeRuntimeReader<Read extends Json> extends BaseReader<Read> {
       sender: ChromeMessageSender,
     ) => void;
 
-    super(async () => {
-      runtime.onMessage.removeListener(messageListener);
-      await onEnd?.();
+    super({
+      onEnd: async () => {
+        runtime.onMessage.removeListener(messageListener);
+        await onEnd?.();
+      },
     });
 
     this.#receiveInput = super.getReceiveInput();
diff --git a/packages/streams/src/MessagePortStream.ts b/packages/streams/src/MessagePortStream.ts
index c4268bbff..fd7360e0f 100644
--- a/packages/streams/src/MessagePortStream.ts
+++ b/packages/streams/src/MessagePortStream.ts
@@ -42,10 +42,12 @@ export class MessagePortReader<Read extends Json> extends BaseReader<Read> {
     // eslint-disable-next-line prefer-const
     let onMessage: OnMessage;
 
-    super(async () => {
-      port.removeEventListener('message', onMessage);
-      port.close();
-      await onEnd?.();
+    super({
+      onEnd: async () => {
+        port.removeEventListener('message', onMessage);
+        port.close();
+        await onEnd?.();
+      },
     });
 
     const receiveInput = super.getReceiveInput();
diff --git a/packages/streams/src/PostMessageStream.ts b/packages/streams/src/PostMessageStream.ts
index 6db292179..4e8e26f16 100644
--- a/packages/streams/src/PostMessageStream.ts
+++ b/packages/streams/src/PostMessageStream.ts
@@ -35,9 +35,11 @@ export class PostMessageReader<Read extends Json> extends BaseReader<Read> {
     // eslint-disable-next-line prefer-const
     let onMessage: OnMessage;
 
-    super(async () => {
-      removeListener(onMessage);
-      await onEnd?.();
+    super({
+      onEnd: async () => {
+        removeListener(onMessage);
+        await onEnd?.();
+      },
     });
 
     const receiveInput = super.getReceiveInput();
diff --git a/packages/streams/test/stream-mocks.ts b/packages/streams/test/stream-mocks.ts
index 4d0a6371b..c98a11bb2 100644
--- a/packages/streams/test/stream-mocks.ts
+++ b/packages/streams/test/stream-mocks.ts
@@ -1,7 +1,12 @@
 import type { Json } from '@metamask/utils';
 
 import { BaseDuplexStream, makeAck } from '../src/BaseDuplexStream.js';
-import type { Dispatch, ReceiveInput } from '../src/BaseStream.js';
+import type {
+  Dispatch,
+  ReceiveInput,
+  BaseReaderArgs,
+  InputValidator,
+} from '../src/BaseStream.js';
 import { BaseReader, BaseWriter } from '../src/BaseStream.js';
 
 export class TestReader<Read extends Json = number> extends BaseReader<Read> {
@@ -11,8 +16,8 @@ export class TestReader<Read extends Json = number> extends BaseReader<Read> {
     return this.#receiveInput;
   }
 
-  constructor(onEnd?: () => void) {
-    super(onEnd);
+  constructor(args: BaseReaderArgs<Read> = {}) {
+    super(args);
     this.#receiveInput = super.getReceiveInput();
   }
 
@@ -34,7 +39,8 @@ export class TestWriter<Write extends Json = number> extends BaseWriter<Write> {
   }
 }
 
-type TestDuplexStreamOptions = {
+type TestDuplexStreamOptions<Read extends Json = number> = {
+  inputValidator?: InputValidator<Read> | undefined;
   readerOnEnd?: () => void;
   writerOnEnd?: () => void;
 };
@@ -57,9 +63,13 @@ export class TestDuplexStream<
 
   constructor(
     onDispatch: Dispatch<Write>,
-    { readerOnEnd, writerOnEnd }: TestDuplexStreamOptions = {},
+    {
+      inputValidator,
+      readerOnEnd,
+      writerOnEnd,
+    }: TestDuplexStreamOptions<Read> = {},
   ) {
-    const reader = new TestReader<Read>(readerOnEnd);
+    const reader = new TestReader<Read>({ inputValidator, onEnd: readerOnEnd });
     super(reader, new TestWriter<Write>(onDispatch, writerOnEnd));
     this.#onDispatch = onDispatch;
     this.#receiveInput = reader.receiveInput;

From 2db0cefd330344c3ec631d9547f8b39dc73ae213 Mon Sep 17 00:00:00 2001
From: Erik Marks <rekmarks@protonmail.com>
Date: Fri, 25 Oct 2024 11:00:00 -0700
Subject: [PATCH 2/4] feat(streams): Thread inputValidator arg through concrete
 stream constructors

---
 .../streams/src/ChromeRuntimeStream.test.ts   | 35 +++++++++++++-
 packages/streams/src/ChromeRuntimeStream.ts   | 20 ++++++--
 .../streams/src/MessagePortStream.test.ts     | 33 ++++++++++++-
 packages/streams/src/MessagePortStream.ts     | 26 +++++++---
 .../streams/src/PostMessageStream.test.ts     | 47 +++++++++++++++++--
 packages/streams/src/PostMessageStream.ts     | 17 ++++---
 6 files changed, 152 insertions(+), 26 deletions(-)

diff --git a/packages/streams/src/ChromeRuntimeStream.test.ts b/packages/streams/src/ChromeRuntimeStream.test.ts
index eaa949691..90dca089b 100644
--- a/packages/streams/src/ChromeRuntimeStream.test.ts
+++ b/packages/streams/src/ChromeRuntimeStream.test.ts
@@ -3,6 +3,7 @@ import { stringify } from '@ocap/utils';
 import { describe, expect, it, vi } from 'vitest';
 
 import { makeAck } from './BaseDuplexStream.js';
+import type { InputValidator } from './BaseStream.js';
 import type { ChromeRuntime } from './chrome.js';
 import type { MessageEnvelope } from './ChromeRuntimeStream.js';
 import {
@@ -94,6 +95,23 @@ describe('ChromeRuntimeReader', () => {
     expect(await reader.next()).toStrictEqual(makePendingResult(message));
   });
 
+  it('calls inputValidator with received input if specified', async () => {
+    const { runtime, dispatchRuntimeMessage } = makeRuntime();
+    const inputValidator = vi.fn();
+    const reader = new ChromeRuntimeReader(
+      asChromeRuntime(runtime),
+      ChromeRuntimeStreamTarget.Background,
+      ChromeRuntimeStreamTarget.Offscreen,
+      { inputValidator },
+    );
+
+    const message = { foo: 'bar' };
+    dispatchRuntimeMessage(message);
+
+    expect(await reader.next()).toStrictEqual(makePendingResult(message));
+    expect(inputValidator).toHaveBeenCalledWith(message);
+  });
+
   it('ignores messages from other extensions', async () => {
     const { runtime, dispatchRuntimeMessage } = makeRuntime();
     const reader = new ChromeRuntimeReader(
@@ -187,7 +205,7 @@ describe('ChromeRuntimeReader', () => {
       asChromeRuntime(runtime),
       ChromeRuntimeStreamTarget.Background,
       ChromeRuntimeStreamTarget.Offscreen,
-      onEnd,
+      { onEnd },
     );
 
     dispatchRuntimeMessage(makeStreamDoneSignal());
@@ -251,12 +269,13 @@ describe.concurrent('ChromeRuntimeWriter', () => {
 
 describe.concurrent('ChromeRuntimeDuplexStream', () => {
   // eslint-disable-next-line @typescript-eslint/explicit-function-return-type
-  const makeDuplexStream = async () => {
+  const makeDuplexStream = async (inputValidator?: InputValidator<number>) => {
     const { runtime, dispatchRuntimeMessage } = makeRuntime();
     const duplexStreamP = ChromeRuntimeDuplexStream.make(
       asChromeRuntime(runtime),
       ChromeRuntimeStreamTarget.Background,
       ChromeRuntimeStreamTarget.Offscreen,
+      inputValidator,
     );
     dispatchRuntimeMessage(makeAck());
 
@@ -282,6 +301,18 @@ describe.concurrent('ChromeRuntimeDuplexStream', () => {
     expect(duplexStream[Symbol.asyncIterator]()).toBe(duplexStream);
   });
 
+  it('calls inputValidator with received input if specified', async () => {
+    const inputValidator = vi.fn();
+    const [duplexStream, { dispatchRuntimeMessage }] =
+      await makeDuplexStream(inputValidator);
+
+    const message = { foo: 'bar' };
+    dispatchRuntimeMessage(message);
+
+    expect(await duplexStream.next()).toStrictEqual(makePendingResult(message));
+    expect(inputValidator).toHaveBeenCalledWith(message);
+  });
+
   it('ends the reader when the writer ends', async () => {
     const [duplexStream, { runtime }] = await makeDuplexStream();
     runtime.sendMessage.mockImplementationOnce(() => {
diff --git a/packages/streams/src/ChromeRuntimeStream.ts b/packages/streams/src/ChromeRuntimeStream.ts
index 0575feeb8..5a0777ec3 100644
--- a/packages/streams/src/ChromeRuntimeStream.ts
+++ b/packages/streams/src/ChromeRuntimeStream.ts
@@ -18,7 +18,12 @@ import type { Json } from '@metamask/utils';
 import { stringify } from '@ocap/utils';
 
 import { BaseDuplexStream } from './BaseDuplexStream.js';
-import type { OnEnd, ReceiveInput } from './BaseStream.js';
+import type {
+  BaseReaderArgs,
+  InputValidator,
+  OnEnd,
+  ReceiveInput,
+} from './BaseStream.js';
 import { BaseReader, BaseWriter } from './BaseStream.js';
 import type { ChromeRuntime, ChromeMessageSender } from './chrome.js';
 import type { Dispatchable } from './utils.js';
@@ -67,7 +72,7 @@ export class ChromeRuntimeReader<Read extends Json> extends BaseReader<Read> {
     runtime: ChromeRuntime,
     target: ChromeRuntimeStreamTarget,
     source: ChromeRuntimeStreamTarget,
-    onEnd?: OnEnd,
+    { inputValidator, onEnd }: BaseReaderArgs<Read> = {},
   ) {
     // eslint-disable-next-line prefer-const
     let messageListener: (
@@ -76,6 +81,7 @@ export class ChromeRuntimeReader<Read extends Json> extends BaseReader<Read> {
     ) => void;
 
     super({
+      inputValidator,
       onEnd: async () => {
         runtime.onMessage.removeListener(messageListener);
         await onEnd?.();
@@ -178,14 +184,18 @@ export class ChromeRuntimeDuplexStream<
     runtime: ChromeRuntime,
     localTarget: ChromeRuntimeStreamTarget,
     remoteTarget: ChromeRuntimeStreamTarget,
+    inputValidator?: InputValidator<Read>,
   ) {
     let writer: ChromeRuntimeWriter<Write>; // eslint-disable-line prefer-const
     const reader = new ChromeRuntimeReader<Read>(
       runtime,
       localTarget,
       remoteTarget,
-      async () => {
-        await writer.return();
+      {
+        inputValidator,
+        onEnd: async () => {
+          await writer.return();
+        },
       },
     );
     writer = new ChromeRuntimeWriter<Write>(
@@ -204,6 +214,7 @@ export class ChromeRuntimeDuplexStream<
     runtime: ChromeRuntime,
     localTarget: ChromeRuntimeStreamTarget,
     remoteTarget: ChromeRuntimeStreamTarget,
+    inputValidator?: InputValidator<Read>,
   ): Promise<ChromeRuntimeDuplexStream<Read, Write>> {
     if (localTarget === remoteTarget) {
       throw new Error('localTarget and remoteTarget must be different');
@@ -213,6 +224,7 @@ export class ChromeRuntimeDuplexStream<
       runtime,
       localTarget,
       remoteTarget,
+      inputValidator,
     );
     await stream.synchronize();
     return stream;
diff --git a/packages/streams/src/MessagePortStream.test.ts b/packages/streams/src/MessagePortStream.test.ts
index 580c40a9f..4a5e9e3b1 100644
--- a/packages/streams/src/MessagePortStream.test.ts
+++ b/packages/streams/src/MessagePortStream.test.ts
@@ -2,6 +2,7 @@ import { delay } from '@ocap/test-utils';
 import { describe, expect, it, vi } from 'vitest';
 
 import { makeAck } from './BaseDuplexStream.js';
+import type { InputValidator } from './BaseStream.js';
 import {
   MessagePortDuplexStream,
   MessagePortReader,
@@ -35,6 +36,19 @@ describe('MessagePortReader', () => {
     expect(await reader.next()).toStrictEqual(makePendingResult(message));
   });
 
+  it('calls inputValidator with received input if specified', async () => {
+    const inputValidator = vi.fn();
+    const { port1, port2 } = new MessageChannel();
+    const reader = new MessagePortReader(port1, { inputValidator });
+
+    const message = { foo: 'bar' };
+    port2.postMessage(message);
+    await delay(10);
+
+    expect(await reader.next()).toStrictEqual(makePendingResult(message));
+    expect(inputValidator).toHaveBeenCalledWith(message);
+  });
+
   it('closes the port when done', async () => {
     const { port1, port2 } = new MessageChannel();
     const closeSpy = vi.spyOn(port1, 'close');
@@ -68,7 +82,7 @@ describe('MessagePortReader', () => {
   it('calls onEnd once when ending', async () => {
     const { port1, port2 } = new MessageChannel();
     const onEnd = vi.fn();
-    const reader = new MessagePortReader(port1, onEnd);
+    const reader = new MessagePortReader(port1, { onEnd });
 
     port2.postMessage(makeStreamDoneSignal());
     await delay(10);
@@ -127,8 +141,12 @@ describe('MessagePortWriter', () => {
 describe('MessagePortDuplexStream', () => {
   const makeDuplexStream = async (
     channel: MessageChannel = new MessageChannel(),
+    inputValidator?: InputValidator<number>,
   ): Promise<MessagePortDuplexStream<number>> => {
-    const duplexStreamP = MessagePortDuplexStream.make<number>(channel.port1);
+    const duplexStreamP = MessagePortDuplexStream.make<number>(
+      channel.port1,
+      inputValidator,
+    );
     channel.port2.postMessage(makeAck());
     await delay(10);
 
@@ -142,6 +160,17 @@ describe('MessagePortDuplexStream', () => {
     expect(duplexStream[Symbol.asyncIterator]()).toBe(duplexStream);
   });
 
+  it('calls inputValidator with received input if specified', async () => {
+    const inputValidator = vi.fn();
+    const channel = new MessageChannel();
+    const duplexStream = await makeDuplexStream(channel, inputValidator);
+
+    channel.port2.postMessage(42);
+
+    expect(await duplexStream.next()).toStrictEqual(makePendingResult(42));
+    expect(inputValidator).toHaveBeenCalledWith(42);
+  });
+
   it('ends the reader when the writer ends', async () => {
     const { port1, port2 } = new MessageChannel();
     vi.spyOn(port1, 'postMessage')
diff --git a/packages/streams/src/MessagePortStream.ts b/packages/streams/src/MessagePortStream.ts
index fd7360e0f..7a33d293c 100644
--- a/packages/streams/src/MessagePortStream.ts
+++ b/packages/streams/src/MessagePortStream.ts
@@ -22,7 +22,7 @@
 import type { Json } from '@metamask/utils';
 
 import { BaseDuplexStream } from './BaseDuplexStream.js';
-import type { OnEnd } from './BaseStream.js';
+import type { BaseReaderArgs, InputValidator, OnEnd } from './BaseStream.js';
 import { BaseReader, BaseWriter } from './BaseStream.js';
 import type { Dispatchable, OnMessage } from './utils.js';
 
@@ -38,11 +38,15 @@ import type { Dispatchable, OnMessage } from './utils.js';
  * - The module-level documentation for more details.
  */
 export class MessagePortReader<Read extends Json> extends BaseReader<Read> {
-  constructor(port: MessagePort, onEnd?: OnEnd) {
+  constructor(
+    port: MessagePort,
+    { inputValidator, onEnd }: BaseReaderArgs<Read> = {},
+  ) {
     // eslint-disable-next-line prefer-const
     let onMessage: OnMessage;
 
     super({
+      inputValidator,
       onEnd: async () => {
         port.removeEventListener('message', onMessage);
         port.close();
@@ -101,10 +105,16 @@ export class MessagePortDuplexStream<
 > {
   // Unavoidable exception to our preference for #-private names.
   // eslint-disable-next-line no-restricted-syntax
-  private constructor(port: MessagePort) {
+  private constructor(
+    port: MessagePort,
+    inputValidator?: InputValidator<Read>,
+  ) {
     let writer: MessagePortWriter<Write>; // eslint-disable-line prefer-const
-    const reader = new MessagePortReader<Read>(port, async () => {
-      await writer.return();
+    const reader = new MessagePortReader<Read>(port, {
+      inputValidator,
+      onEnd: async () => {
+        await writer.return();
+      },
     });
     writer = new MessagePortWriter<Write>(port, async () => {
       await reader.return();
@@ -114,8 +124,12 @@ export class MessagePortDuplexStream<
 
   static async make<Read extends Json, Write extends Json = Read>(
     port: MessagePort,
+    inputValidator?: InputValidator<Read>,
   ): Promise<MessagePortDuplexStream<Read, Write>> {
-    const stream = new MessagePortDuplexStream<Read, Write>(port);
+    const stream = new MessagePortDuplexStream<Read, Write>(
+      port,
+      inputValidator,
+    );
     await stream.synchronize();
     return stream;
   }
diff --git a/packages/streams/src/PostMessageStream.test.ts b/packages/streams/src/PostMessageStream.test.ts
index c85d2d088..896990ed5 100644
--- a/packages/streams/src/PostMessageStream.test.ts
+++ b/packages/streams/src/PostMessageStream.test.ts
@@ -2,6 +2,7 @@ import { delay } from '@ocap/test-utils';
 import { describe, it, expect, vi } from 'vitest';
 
 import { makeAck } from './BaseDuplexStream.js';
+import type { InputValidator } from './BaseStream.js';
 import {
   PostMessageDuplexStream,
   PostMessageReader,
@@ -50,6 +51,20 @@ describe('PostMessageReader', () => {
     expect(await reader.next()).toStrictEqual(makePendingResult(message));
   });
 
+  it('calls inputValidator with received input if specified', async () => {
+    const inputValidator = vi.fn();
+    const { postMessageFn, setListener, removeListener } =
+      makePostMessageMock();
+    const reader = new PostMessageReader(setListener, removeListener, {
+      inputValidator,
+    });
+
+    const message = { foo: 'bar' };
+    postMessageFn(message);
+    expect(await reader.next()).toStrictEqual(makePendingResult(message));
+    expect(inputValidator).toHaveBeenCalledWith(message);
+  });
+
   it('removes its listener when it ends', async () => {
     const { postMessageFn, setListener, removeListener, listeners } =
       makePostMessageMock();
@@ -83,7 +98,9 @@ describe('PostMessageReader', () => {
     const { postMessageFn, setListener, removeListener } =
       makePostMessageMock();
     const onEnd = vi.fn();
-    const reader = new PostMessageReader(setListener, removeListener, onEnd);
+    const reader = new PostMessageReader(setListener, removeListener, {
+      onEnd,
+    });
 
     postMessageFn(makeStreamDoneSignal());
 
@@ -121,15 +138,21 @@ describe('PostMessageWriter', () => {
 });
 
 describe('PostMessageDuplexStream', () => {
-  // eslint-disable-next-line @typescript-eslint/explicit-function-return-type
-  const makeDuplexStream = async (sendMessage: PostMessage) => {
-    const { postMessageFn, setListener, removeListener } =
-      makePostMessageMock();
+  const makeDuplexStream = async (
+    sendMessage: PostMessage,
+    postMessageMock: ReturnType<
+      typeof makePostMessageMock
+    > = makePostMessageMock(),
+    inputValidator?: InputValidator<number>,
+    // eslint-disable-next-line @typescript-eslint/explicit-function-return-type
+  ) => {
+    const { postMessageFn, setListener, removeListener } = postMessageMock;
 
     const duplexStreamP = PostMessageDuplexStream.make(
       sendMessage,
       setListener,
       removeListener,
+      inputValidator,
     );
     postMessageFn(makeAck());
     await delay(10);
@@ -144,6 +167,20 @@ describe('PostMessageDuplexStream', () => {
     expect(duplexStream[Symbol.asyncIterator]()).toBe(duplexStream);
   });
 
+  it('calls inputValidator with received input if specified', async () => {
+    const inputValidator = vi.fn();
+    const postMessageMock = makePostMessageMock();
+    const [duplexStream] = await makeDuplexStream(
+      () => undefined,
+      postMessageMock,
+      inputValidator,
+    );
+
+    postMessageMock.postMessageFn(42);
+    expect(await duplexStream.next()).toStrictEqual(makePendingResult(42));
+    expect(inputValidator).toHaveBeenCalledWith(42);
+  });
+
   it('ends the reader when the writer ends', async () => {
     const [duplexStream] = await makeDuplexStream(
       vi
diff --git a/packages/streams/src/PostMessageStream.ts b/packages/streams/src/PostMessageStream.ts
index 4e8e26f16..b15f115fe 100644
--- a/packages/streams/src/PostMessageStream.ts
+++ b/packages/streams/src/PostMessageStream.ts
@@ -9,7 +9,7 @@
 import type { Json } from '@metamask/utils';
 
 import { BaseDuplexStream } from './BaseDuplexStream.js';
-import type { OnEnd } from './BaseStream.js';
+import type { BaseReaderArgs, InputValidator, OnEnd } from './BaseStream.js';
 import { BaseReader, BaseWriter } from './BaseStream.js';
 // Used in docstring.
 // eslint-disable-next-line @typescript-eslint/no-unused-vars
@@ -30,12 +30,13 @@ export class PostMessageReader<Read extends Json> extends BaseReader<Read> {
   constructor(
     setListener: SetListener,
     removeListener: RemoveListener,
-    onEnd?: OnEnd,
+    { inputValidator, onEnd }: BaseReaderArgs<Read> = {},
   ) {
     // eslint-disable-next-line prefer-const
     let onMessage: OnMessage;
 
     super({
+      inputValidator,
       onEnd: async () => {
         removeListener(onMessage);
         await onEnd?.();
@@ -91,15 +92,15 @@ export class PostMessageDuplexStream<
     postMessageFn: PostMessage,
     setListener: SetListener,
     removeListener: RemoveListener,
+    inputValidator?: InputValidator<Read>,
   ) {
     let writer: PostMessageWriter<Write>; // eslint-disable-line prefer-const
-    const reader = new PostMessageReader<Read>(
-      setListener,
-      removeListener,
-      async () => {
+    const reader = new PostMessageReader<Read>(setListener, removeListener, {
+      inputValidator,
+      onEnd: async () => {
         await writer.return();
       },
-    );
+    });
     writer = new PostMessageWriter<Write>(postMessageFn, async () => {
       await reader.return();
     });
@@ -110,11 +111,13 @@ export class PostMessageDuplexStream<
     postMessageFn: PostMessage,
     setListener: SetListener,
     removeListener: RemoveListener,
+    inputValidator?: InputValidator<Read>,
   ): Promise<PostMessageDuplexStream<Read, Write>> {
     const stream = new PostMessageDuplexStream<Read, Write>(
       postMessageFn,
       setListener,
       removeListener,
+      inputValidator,
     );
     await stream.synchronize();
     return stream;

From 109ddcacd93b49ab039e6444e947af1ee27b3d87 Mon Sep 17 00:00:00 2001
From: Erik Marks <rekmarks@protonmail.com>
Date: Tue, 29 Oct 2024 22:25:33 -0700
Subject: [PATCH 3/4] feat(streams): Make input validation failure end the
 stream

---
 packages/streams/src/BaseStream.test.ts       | 50 ++++++++++++----
 packages/streams/src/BaseStream.ts            | 57 +++++++++++--------
 .../streams/src/ChromeRuntimeStream.test.ts   | 26 +++++----
 packages/streams/src/ChromeRuntimeStream.ts   | 14 ++---
 .../streams/src/MessagePortStream.test.ts     | 26 +++++----
 packages/streams/src/MessagePortStream.ts     | 17 +++---
 .../streams/src/PostMessageStream.test.ts     | 26 +++++----
 packages/streams/src/PostMessageStream.ts     | 14 ++---
 packages/streams/test/stream-mocks.ts         | 10 ++--
 9 files changed, 142 insertions(+), 98 deletions(-)

diff --git a/packages/streams/src/BaseStream.test.ts b/packages/streams/src/BaseStream.test.ts
index 51f60408b..dbf1e16b5 100644
--- a/packages/streams/src/BaseStream.test.ts
+++ b/packages/streams/src/BaseStream.test.ts
@@ -2,6 +2,7 @@ import { makeErrorMatcherFactory } from '@ocap/test-utils';
 import { describe, expect, it, vi } from 'vitest';
 
 import { BaseReader, BaseWriter } from './BaseStream.js';
+import type { ValidateInput } from './BaseStream.js';
 import {
   makeDoneResult,
   makePendingResult,
@@ -50,15 +51,17 @@ describe('BaseReader', () => {
       expect(await reader.next()).toStrictEqual(makePendingResult(message));
     });
 
-    it('calls inputValidator with received input if specified', async () => {
-      const inputValidator = vi.fn();
-      const reader = new TestReader({ inputValidator });
+    it('calls validateInput with received input if specified', async () => {
+      const validateInput = vi.fn((_value: unknown) => true);
+      const reader = new TestReader({
+        validateInput: validateInput as unknown as ValidateInput<number>,
+      });
 
       const message = 42;
       reader.receiveInput(message);
 
       expect(await reader.next()).toStrictEqual(makePendingResult(message));
-      expect(inputValidator).toHaveBeenCalledWith(message);
+      expect(validateInput).toHaveBeenCalledWith(message);
     });
 
     it('emits message received after next()', async () => {
@@ -88,26 +91,26 @@ describe('BaseReader', () => {
       }
     });
 
-    it('throws after receiving unexpected message, before read is enqueued', async () => {
+    it('throws after receiving non-Dispatchable input, before read is enqueued', async () => {
       const reader = new TestReader();
 
-      const unexpectedMessage = Symbol('foo');
-      reader.receiveInput(unexpectedMessage);
+      const badMessage = Symbol('foo');
+      reader.receiveInput(badMessage);
 
       await expect(reader.next()).rejects.toThrow(
-        'Received unexpected message from transport',
+        'Message cannot be processed by stream (must be JSON-serializable)',
       );
     });
 
-    it('throws after receiving unexpected message, after read is enqueued', async () => {
+    it('throws after receiving non-Dispatchable input, after read is enqueued', async () => {
       const reader = new TestReader();
 
       const nextP = reader.next();
-      const unexpectedMessage = Symbol('foo');
-      reader.receiveInput(unexpectedMessage);
+      const badMessage = Symbol('foo');
+      reader.receiveInput(badMessage);
 
       await expect(nextP).rejects.toThrow(
-        'Received unexpected message from transport',
+        'Message cannot be processed by stream (must be JSON-serializable)',
       );
     });
 
@@ -128,6 +131,29 @@ describe('BaseReader', () => {
       await expect(nextP).rejects.toThrow('foo');
     });
 
+    it('throws after receiving invalid input, before read is enqueued', async () => {
+      const reader = new TestReader({
+        validateInput: (value) => typeof value === 'number',
+      });
+
+      reader.receiveInput({});
+
+      await expect(reader.next()).rejects.toThrow(
+        'Message failed type validation',
+      );
+    });
+
+    it('throws after receiving invalid input, after read is enqueued', async () => {
+      const reader = new TestReader({
+        validateInput: (value) => typeof value === 'number',
+      });
+
+      const nextP = reader.next();
+      reader.receiveInput({});
+
+      await expect(nextP).rejects.toThrow('Message failed type validation');
+    });
+
     it('ends after receiving done signal, before read is enqueued', async () => {
       const reader = new TestReader();
 
diff --git a/packages/streams/src/BaseStream.ts b/packages/streams/src/BaseStream.ts
index ee68e46b5..9de32e278 100644
--- a/packages/streams/src/BaseStream.ts
+++ b/packages/streams/src/BaseStream.ts
@@ -94,14 +94,10 @@ harden(makeStreamBuffer);
  */
 export type OnEnd = () => void | Promise<void>;
 
-export type InputValidator<Read extends Json> = (
-  input: Json,
-) => asserts input is Read;
-
-export type BaseReaderArgs<Read extends Json> = {
-  inputValidator?: InputValidator<Read> | undefined;
-  onEnd?: OnEnd | undefined;
-};
+/**
+ * A function that validates input to a readable stream.
+ */
+export type ValidateInput<Read extends Json> = (input: Json) => input is Read;
 
 /**
  * A function that receives input from a transport mechanism to a readable stream.
@@ -109,6 +105,11 @@ export type BaseReaderArgs<Read extends Json> = {
  */
 export type ReceiveInput = (input: unknown) => void;
 
+export type BaseReaderArgs<Read extends Json> = {
+  validateInput?: ValidateInput<Read> | undefined;
+  onEnd?: OnEnd | undefined;
+};
+
 /**
  * The base of a readable async iterator stream.
  *
@@ -126,7 +127,7 @@ export class BaseReader<Read extends Json> implements Reader<Read> {
    */
   readonly #buffer = makeStreamBuffer<IteratorResult<Read, undefined>>();
 
-  readonly #inputValidator?: InputValidator<Read> | undefined;
+  readonly #validateInput?: ValidateInput<Read> | undefined;
 
   #onEnd?: OnEnd | undefined;
 
@@ -136,12 +137,12 @@ export class BaseReader<Read extends Json> implements Reader<Read> {
    * Constructs a {@link BaseReader}.
    *
    * @param args - Options bag.
-   * @param args.inputValidator - A function that validates input from the transport.
+   * @param args.validateInput - A function that validates input from the transport.
    * @param args.onEnd - A function that is called when the stream ends. For any cleanup that
    * should happen when the stream ends, such as closing a message port.
    */
-  constructor({ inputValidator, onEnd }: BaseReaderArgs<Read>) {
-    this.#inputValidator = inputValidator;
+  constructor({ validateInput, onEnd }: BaseReaderArgs<Read>) {
+    this.#validateInput = validateInput;
     this.#onEnd = onEnd;
     harden(this);
   }
@@ -162,22 +163,17 @@ export class BaseReader<Read extends Json> implements Reader<Read> {
 
   readonly #receiveInput: ReceiveInput = async (input) => {
     if (!isDispatchable(input)) {
-      const error = new Error(
-        `Received unexpected message from transport:\n${stringify(input)}`,
+      await this.#handleInputError(
+        new Error(
+          `Message cannot be processed by stream (must be JSON-serializable):\n${stringify(input)}`,
+        ),
       );
-      if (!this.#buffer.hasPendingReads()) {
-        this.#buffer.put(error);
-      }
-      await this.#end(error);
       return;
     }
 
     const unmarshaled = unmarshal(input);
     if (unmarshaled instanceof Error) {
-      if (!this.#buffer.hasPendingReads()) {
-        this.#buffer.put(unmarshaled);
-      }
-      await this.#end(unmarshaled);
+      await this.#handleInputError(unmarshaled);
       return;
     }
 
@@ -186,10 +182,23 @@ export class BaseReader<Read extends Json> implements Reader<Read> {
       return;
     }
 
-    this.#inputValidator?.(unmarshaled);
-    this.#buffer.put(makePendingResult(unmarshaled as Read));
+    if (this.#validateInput?.(unmarshaled) === false) {
+      await this.#handleInputError(
+        new Error(`Message failed type validation:\n${stringify(unmarshaled)}`),
+      );
+      return;
+    }
+
+    this.#buffer.put(makePendingResult(unmarshaled));
   };
 
+  async #handleInputError(error: Error): Promise<void> {
+    if (!this.#buffer.hasPendingReads()) {
+      this.#buffer.put(error);
+    }
+    await this.#end(error);
+  }
+
   /**
    * Ends the stream. Calls and then unsets the `#onEnd` method.
    * Idempotent.
diff --git a/packages/streams/src/ChromeRuntimeStream.test.ts b/packages/streams/src/ChromeRuntimeStream.test.ts
index 90dca089b..d208f77c8 100644
--- a/packages/streams/src/ChromeRuntimeStream.test.ts
+++ b/packages/streams/src/ChromeRuntimeStream.test.ts
@@ -3,7 +3,7 @@ import { stringify } from '@ocap/utils';
 import { describe, expect, it, vi } from 'vitest';
 
 import { makeAck } from './BaseDuplexStream.js';
-import type { InputValidator } from './BaseStream.js';
+import type { ValidateInput } from './BaseStream.js';
 import type { ChromeRuntime } from './chrome.js';
 import type { MessageEnvelope } from './ChromeRuntimeStream.js';
 import {
@@ -95,21 +95,23 @@ describe('ChromeRuntimeReader', () => {
     expect(await reader.next()).toStrictEqual(makePendingResult(message));
   });
 
-  it('calls inputValidator with received input if specified', async () => {
+  it('calls validateInput with received input if specified', async () => {
     const { runtime, dispatchRuntimeMessage } = makeRuntime();
-    const inputValidator = vi.fn();
+    const validateInput = vi
+      .fn()
+      .mockReturnValue(true) as unknown as ValidateInput<number>;
     const reader = new ChromeRuntimeReader(
       asChromeRuntime(runtime),
       ChromeRuntimeStreamTarget.Background,
       ChromeRuntimeStreamTarget.Offscreen,
-      { inputValidator },
+      { validateInput },
     );
 
     const message = { foo: 'bar' };
     dispatchRuntimeMessage(message);
 
     expect(await reader.next()).toStrictEqual(makePendingResult(message));
-    expect(inputValidator).toHaveBeenCalledWith(message);
+    expect(validateInput).toHaveBeenCalledWith(message);
   });
 
   it('ignores messages from other extensions', async () => {
@@ -269,13 +271,13 @@ describe.concurrent('ChromeRuntimeWriter', () => {
 
 describe.concurrent('ChromeRuntimeDuplexStream', () => {
   // eslint-disable-next-line @typescript-eslint/explicit-function-return-type
-  const makeDuplexStream = async (inputValidator?: InputValidator<number>) => {
+  const makeDuplexStream = async (validateInput?: ValidateInput<number>) => {
     const { runtime, dispatchRuntimeMessage } = makeRuntime();
     const duplexStreamP = ChromeRuntimeDuplexStream.make(
       asChromeRuntime(runtime),
       ChromeRuntimeStreamTarget.Background,
       ChromeRuntimeStreamTarget.Offscreen,
-      inputValidator,
+      validateInput,
     );
     dispatchRuntimeMessage(makeAck());
 
@@ -301,16 +303,18 @@ describe.concurrent('ChromeRuntimeDuplexStream', () => {
     expect(duplexStream[Symbol.asyncIterator]()).toBe(duplexStream);
   });
 
-  it('calls inputValidator with received input if specified', async () => {
-    const inputValidator = vi.fn();
+  it('calls validateInput with received input if specified', async () => {
+    const validateInput = vi
+      .fn()
+      .mockReturnValue(true) as unknown as ValidateInput<number>;
     const [duplexStream, { dispatchRuntimeMessage }] =
-      await makeDuplexStream(inputValidator);
+      await makeDuplexStream(validateInput);
 
     const message = { foo: 'bar' };
     dispatchRuntimeMessage(message);
 
     expect(await duplexStream.next()).toStrictEqual(makePendingResult(message));
-    expect(inputValidator).toHaveBeenCalledWith(message);
+    expect(validateInput).toHaveBeenCalledWith(message);
   });
 
   it('ends the reader when the writer ends', async () => {
diff --git a/packages/streams/src/ChromeRuntimeStream.ts b/packages/streams/src/ChromeRuntimeStream.ts
index 5a0777ec3..ee1a7483d 100644
--- a/packages/streams/src/ChromeRuntimeStream.ts
+++ b/packages/streams/src/ChromeRuntimeStream.ts
@@ -20,7 +20,7 @@ import { stringify } from '@ocap/utils';
 import { BaseDuplexStream } from './BaseDuplexStream.js';
 import type {
   BaseReaderArgs,
-  InputValidator,
+  ValidateInput,
   OnEnd,
   ReceiveInput,
 } from './BaseStream.js';
@@ -72,7 +72,7 @@ export class ChromeRuntimeReader<Read extends Json> extends BaseReader<Read> {
     runtime: ChromeRuntime,
     target: ChromeRuntimeStreamTarget,
     source: ChromeRuntimeStreamTarget,
-    { inputValidator, onEnd }: BaseReaderArgs<Read> = {},
+    { validateInput, onEnd }: BaseReaderArgs<Read> = {},
   ) {
     // eslint-disable-next-line prefer-const
     let messageListener: (
@@ -81,7 +81,7 @@ export class ChromeRuntimeReader<Read extends Json> extends BaseReader<Read> {
     ) => void;
 
     super({
-      inputValidator,
+      validateInput,
       onEnd: async () => {
         runtime.onMessage.removeListener(messageListener);
         await onEnd?.();
@@ -184,7 +184,7 @@ export class ChromeRuntimeDuplexStream<
     runtime: ChromeRuntime,
     localTarget: ChromeRuntimeStreamTarget,
     remoteTarget: ChromeRuntimeStreamTarget,
-    inputValidator?: InputValidator<Read>,
+    validateInput?: ValidateInput<Read>,
   ) {
     let writer: ChromeRuntimeWriter<Write>; // eslint-disable-line prefer-const
     const reader = new ChromeRuntimeReader<Read>(
@@ -192,7 +192,7 @@ export class ChromeRuntimeDuplexStream<
       localTarget,
       remoteTarget,
       {
-        inputValidator,
+        validateInput,
         onEnd: async () => {
           await writer.return();
         },
@@ -214,7 +214,7 @@ export class ChromeRuntimeDuplexStream<
     runtime: ChromeRuntime,
     localTarget: ChromeRuntimeStreamTarget,
     remoteTarget: ChromeRuntimeStreamTarget,
-    inputValidator?: InputValidator<Read>,
+    validateInput?: ValidateInput<Read>,
   ): Promise<ChromeRuntimeDuplexStream<Read, Write>> {
     if (localTarget === remoteTarget) {
       throw new Error('localTarget and remoteTarget must be different');
@@ -224,7 +224,7 @@ export class ChromeRuntimeDuplexStream<
       runtime,
       localTarget,
       remoteTarget,
-      inputValidator,
+      validateInput,
     );
     await stream.synchronize();
     return stream;
diff --git a/packages/streams/src/MessagePortStream.test.ts b/packages/streams/src/MessagePortStream.test.ts
index 4a5e9e3b1..ff6191cf8 100644
--- a/packages/streams/src/MessagePortStream.test.ts
+++ b/packages/streams/src/MessagePortStream.test.ts
@@ -2,7 +2,7 @@ import { delay } from '@ocap/test-utils';
 import { describe, expect, it, vi } from 'vitest';
 
 import { makeAck } from './BaseDuplexStream.js';
-import type { InputValidator } from './BaseStream.js';
+import type { ValidateInput } from './BaseStream.js';
 import {
   MessagePortDuplexStream,
   MessagePortReader,
@@ -36,17 +36,19 @@ describe('MessagePortReader', () => {
     expect(await reader.next()).toStrictEqual(makePendingResult(message));
   });
 
-  it('calls inputValidator with received input if specified', async () => {
-    const inputValidator = vi.fn();
+  it('calls validateInput with received input if specified', async () => {
+    const validateInput = vi
+      .fn()
+      .mockReturnValue(true) as unknown as ValidateInput<number>;
     const { port1, port2 } = new MessageChannel();
-    const reader = new MessagePortReader(port1, { inputValidator });
+    const reader = new MessagePortReader(port1, { validateInput });
 
     const message = { foo: 'bar' };
     port2.postMessage(message);
     await delay(10);
 
     expect(await reader.next()).toStrictEqual(makePendingResult(message));
-    expect(inputValidator).toHaveBeenCalledWith(message);
+    expect(validateInput).toHaveBeenCalledWith(message);
   });
 
   it('closes the port when done', async () => {
@@ -141,11 +143,11 @@ describe('MessagePortWriter', () => {
 describe('MessagePortDuplexStream', () => {
   const makeDuplexStream = async (
     channel: MessageChannel = new MessageChannel(),
-    inputValidator?: InputValidator<number>,
+    validateInput?: ValidateInput<number>,
   ): Promise<MessagePortDuplexStream<number>> => {
     const duplexStreamP = MessagePortDuplexStream.make<number>(
       channel.port1,
-      inputValidator,
+      validateInput,
     );
     channel.port2.postMessage(makeAck());
     await delay(10);
@@ -160,15 +162,17 @@ describe('MessagePortDuplexStream', () => {
     expect(duplexStream[Symbol.asyncIterator]()).toBe(duplexStream);
   });
 
-  it('calls inputValidator with received input if specified', async () => {
-    const inputValidator = vi.fn();
+  it('calls validateInput with received input if specified', async () => {
+    const validateInput = vi
+      .fn()
+      .mockReturnValue(true) as unknown as ValidateInput<number>;
     const channel = new MessageChannel();
-    const duplexStream = await makeDuplexStream(channel, inputValidator);
+    const duplexStream = await makeDuplexStream(channel, validateInput);
 
     channel.port2.postMessage(42);
 
     expect(await duplexStream.next()).toStrictEqual(makePendingResult(42));
-    expect(inputValidator).toHaveBeenCalledWith(42);
+    expect(validateInput).toHaveBeenCalledWith(42);
   });
 
   it('ends the reader when the writer ends', async () => {
diff --git a/packages/streams/src/MessagePortStream.ts b/packages/streams/src/MessagePortStream.ts
index 7a33d293c..e27238628 100644
--- a/packages/streams/src/MessagePortStream.ts
+++ b/packages/streams/src/MessagePortStream.ts
@@ -22,7 +22,7 @@
 import type { Json } from '@metamask/utils';
 
 import { BaseDuplexStream } from './BaseDuplexStream.js';
-import type { BaseReaderArgs, InputValidator, OnEnd } from './BaseStream.js';
+import type { BaseReaderArgs, ValidateInput, OnEnd } from './BaseStream.js';
 import { BaseReader, BaseWriter } from './BaseStream.js';
 import type { Dispatchable, OnMessage } from './utils.js';
 
@@ -40,13 +40,13 @@ import type { Dispatchable, OnMessage } from './utils.js';
 export class MessagePortReader<Read extends Json> extends BaseReader<Read> {
   constructor(
     port: MessagePort,
-    { inputValidator, onEnd }: BaseReaderArgs<Read> = {},
+    { validateInput, onEnd }: BaseReaderArgs<Read> = {},
   ) {
     // eslint-disable-next-line prefer-const
     let onMessage: OnMessage;
 
     super({
-      inputValidator,
+      validateInput,
       onEnd: async () => {
         port.removeEventListener('message', onMessage);
         port.close();
@@ -105,13 +105,10 @@ export class MessagePortDuplexStream<
 > {
   // Unavoidable exception to our preference for #-private names.
   // eslint-disable-next-line no-restricted-syntax
-  private constructor(
-    port: MessagePort,
-    inputValidator?: InputValidator<Read>,
-  ) {
+  private constructor(port: MessagePort, validateInput?: ValidateInput<Read>) {
     let writer: MessagePortWriter<Write>; // eslint-disable-line prefer-const
     const reader = new MessagePortReader<Read>(port, {
-      inputValidator,
+      validateInput,
       onEnd: async () => {
         await writer.return();
       },
@@ -124,11 +121,11 @@ export class MessagePortDuplexStream<
 
   static async make<Read extends Json, Write extends Json = Read>(
     port: MessagePort,
-    inputValidator?: InputValidator<Read>,
+    validateInput?: ValidateInput<Read>,
   ): Promise<MessagePortDuplexStream<Read, Write>> {
     const stream = new MessagePortDuplexStream<Read, Write>(
       port,
-      inputValidator,
+      validateInput,
     );
     await stream.synchronize();
     return stream;
diff --git a/packages/streams/src/PostMessageStream.test.ts b/packages/streams/src/PostMessageStream.test.ts
index 896990ed5..54e9e2b30 100644
--- a/packages/streams/src/PostMessageStream.test.ts
+++ b/packages/streams/src/PostMessageStream.test.ts
@@ -2,7 +2,7 @@ import { delay } from '@ocap/test-utils';
 import { describe, it, expect, vi } from 'vitest';
 
 import { makeAck } from './BaseDuplexStream.js';
-import type { InputValidator } from './BaseStream.js';
+import type { ValidateInput } from './BaseStream.js';
 import {
   PostMessageDuplexStream,
   PostMessageReader,
@@ -51,18 +51,20 @@ describe('PostMessageReader', () => {
     expect(await reader.next()).toStrictEqual(makePendingResult(message));
   });
 
-  it('calls inputValidator with received input if specified', async () => {
-    const inputValidator = vi.fn();
+  it('calls validateInput with received input if specified', async () => {
+    const validateInput = vi
+      .fn()
+      .mockReturnValue(true) as unknown as ValidateInput<number>;
     const { postMessageFn, setListener, removeListener } =
       makePostMessageMock();
     const reader = new PostMessageReader(setListener, removeListener, {
-      inputValidator,
+      validateInput,
     });
 
     const message = { foo: 'bar' };
     postMessageFn(message);
     expect(await reader.next()).toStrictEqual(makePendingResult(message));
-    expect(inputValidator).toHaveBeenCalledWith(message);
+    expect(validateInput).toHaveBeenCalledWith(message);
   });
 
   it('removes its listener when it ends', async () => {
@@ -143,7 +145,7 @@ describe('PostMessageDuplexStream', () => {
     postMessageMock: ReturnType<
       typeof makePostMessageMock
     > = makePostMessageMock(),
-    inputValidator?: InputValidator<number>,
+    validateInput?: ValidateInput<number>,
     // eslint-disable-next-line @typescript-eslint/explicit-function-return-type
   ) => {
     const { postMessageFn, setListener, removeListener } = postMessageMock;
@@ -152,7 +154,7 @@ describe('PostMessageDuplexStream', () => {
       sendMessage,
       setListener,
       removeListener,
-      inputValidator,
+      validateInput,
     );
     postMessageFn(makeAck());
     await delay(10);
@@ -167,18 +169,20 @@ describe('PostMessageDuplexStream', () => {
     expect(duplexStream[Symbol.asyncIterator]()).toBe(duplexStream);
   });
 
-  it('calls inputValidator with received input if specified', async () => {
-    const inputValidator = vi.fn();
+  it('calls validateInput with received input if specified', async () => {
+    const validateInput = vi
+      .fn()
+      .mockReturnValue(true) as unknown as ValidateInput<number>;
     const postMessageMock = makePostMessageMock();
     const [duplexStream] = await makeDuplexStream(
       () => undefined,
       postMessageMock,
-      inputValidator,
+      validateInput,
     );
 
     postMessageMock.postMessageFn(42);
     expect(await duplexStream.next()).toStrictEqual(makePendingResult(42));
-    expect(inputValidator).toHaveBeenCalledWith(42);
+    expect(validateInput).toHaveBeenCalledWith(42);
   });
 
   it('ends the reader when the writer ends', async () => {
diff --git a/packages/streams/src/PostMessageStream.ts b/packages/streams/src/PostMessageStream.ts
index b15f115fe..e70772049 100644
--- a/packages/streams/src/PostMessageStream.ts
+++ b/packages/streams/src/PostMessageStream.ts
@@ -9,7 +9,7 @@
 import type { Json } from '@metamask/utils';
 
 import { BaseDuplexStream } from './BaseDuplexStream.js';
-import type { BaseReaderArgs, InputValidator, OnEnd } from './BaseStream.js';
+import type { BaseReaderArgs, ValidateInput, OnEnd } from './BaseStream.js';
 import { BaseReader, BaseWriter } from './BaseStream.js';
 // Used in docstring.
 // eslint-disable-next-line @typescript-eslint/no-unused-vars
@@ -30,13 +30,13 @@ export class PostMessageReader<Read extends Json> extends BaseReader<Read> {
   constructor(
     setListener: SetListener,
     removeListener: RemoveListener,
-    { inputValidator, onEnd }: BaseReaderArgs<Read> = {},
+    { validateInput, onEnd }: BaseReaderArgs<Read> = {},
   ) {
     // eslint-disable-next-line prefer-const
     let onMessage: OnMessage;
 
     super({
-      inputValidator,
+      validateInput,
       onEnd: async () => {
         removeListener(onMessage);
         await onEnd?.();
@@ -92,11 +92,11 @@ export class PostMessageDuplexStream<
     postMessageFn: PostMessage,
     setListener: SetListener,
     removeListener: RemoveListener,
-    inputValidator?: InputValidator<Read>,
+    validateInput?: ValidateInput<Read>,
   ) {
     let writer: PostMessageWriter<Write>; // eslint-disable-line prefer-const
     const reader = new PostMessageReader<Read>(setListener, removeListener, {
-      inputValidator,
+      validateInput,
       onEnd: async () => {
         await writer.return();
       },
@@ -111,13 +111,13 @@ export class PostMessageDuplexStream<
     postMessageFn: PostMessage,
     setListener: SetListener,
     removeListener: RemoveListener,
-    inputValidator?: InputValidator<Read>,
+    validateInput?: ValidateInput<Read>,
   ): Promise<PostMessageDuplexStream<Read, Write>> {
     const stream = new PostMessageDuplexStream<Read, Write>(
       postMessageFn,
       setListener,
       removeListener,
-      inputValidator,
+      validateInput,
     );
     await stream.synchronize();
     return stream;
diff --git a/packages/streams/test/stream-mocks.ts b/packages/streams/test/stream-mocks.ts
index c98a11bb2..e917d7318 100644
--- a/packages/streams/test/stream-mocks.ts
+++ b/packages/streams/test/stream-mocks.ts
@@ -5,7 +5,7 @@ import type {
   Dispatch,
   ReceiveInput,
   BaseReaderArgs,
-  InputValidator,
+  ValidateInput,
 } from '../src/BaseStream.js';
 import { BaseReader, BaseWriter } from '../src/BaseStream.js';
 
@@ -40,7 +40,7 @@ export class TestWriter<Write extends Json = number> extends BaseWriter<Write> {
 }
 
 type TestDuplexStreamOptions<Read extends Json = number> = {
-  inputValidator?: InputValidator<Read> | undefined;
+  validateInput?: ValidateInput<Read> | undefined;
   readerOnEnd?: () => void;
   writerOnEnd?: () => void;
 };
@@ -64,12 +64,12 @@ export class TestDuplexStream<
   constructor(
     onDispatch: Dispatch<Write>,
     {
-      inputValidator,
+      validateInput,
       readerOnEnd,
       writerOnEnd,
     }: TestDuplexStreamOptions<Read> = {},
   ) {
-    const reader = new TestReader<Read>({ inputValidator, onEnd: readerOnEnd });
+    const reader = new TestReader<Read>({ validateInput, onEnd: readerOnEnd });
     super(reader, new TestWriter<Write>(onDispatch, writerOnEnd));
     this.#onDispatch = onDispatch;
     this.#receiveInput = reader.receiveInput;
@@ -99,7 +99,7 @@ export class TestDuplexStream<
    */
   static async make<Read extends Json = number, Write extends Json = Read>(
     onDispatch: Dispatch<Write>,
-    opts: TestDuplexStreamOptions = {},
+    opts: TestDuplexStreamOptions<Read> = {},
   ): Promise<TestDuplexStream<Read, Write>> {
     const stream = new TestDuplexStream<Read, Write>(onDispatch, opts);
     await stream.completeSynchronization();

From df79e28c1f8f799a6d73c4c52490b0450600d40f Mon Sep 17 00:00:00 2001
From: Erik Marks <rekmarks@protonmail.com>
Date: Wed, 6 Nov 2024 15:14:48 +0700
Subject: [PATCH 4/4] test(kernel): Fix failing tests

---
 packages/kernel/src/Supervisor.test.ts | 4 +++-
 packages/kernel/src/Vat.test.ts        | 4 +++-
 2 files changed, 6 insertions(+), 2 deletions(-)

diff --git a/packages/kernel/src/Supervisor.test.ts b/packages/kernel/src/Supervisor.test.ts
index 3823831c1..c311ec827 100644
--- a/packages/kernel/src/Supervisor.test.ts
+++ b/packages/kernel/src/Supervisor.test.ts
@@ -34,7 +34,9 @@ describe('Supervisor', () => {
       await delay(10);
       expect(consoleErrorSpy).toHaveBeenCalledWith(
         `Unexpected read error from Supervisor "${supervisor.id}"`,
-        new Error('Received unexpected message from transport:\nnull'),
+        new Error(
+          'Message cannot be processed by stream (must be JSON-serializable):\nnull',
+        ),
       );
     });
   });
diff --git a/packages/kernel/src/Vat.test.ts b/packages/kernel/src/Vat.test.ts
index 998a2f882..d2d73768d 100644
--- a/packages/kernel/src/Vat.test.ts
+++ b/packages/kernel/src/Vat.test.ts
@@ -68,7 +68,9 @@ describe('Vat', () => {
       await delay(10);
       expect(consoleErrorSpy).toHaveBeenCalledWith(
         'Unexpected read error',
-        new Error('Received unexpected message from transport:\nnull'),
+        new Error(
+          'Message cannot be processed by stream (must be JSON-serializable):\nnull',
+        ),
       );
     });
   });