feat(transport): add rate limiting for messages and connections

packages/kernel-errors/src/errors/ResourceLimitError.test.ts

@@ -167,7 +167,7 @@ describe('ResourceLimitError', () => {
         stack: 'stack trace',
       } as unknown as MarshaledOcapError,
       expectedError:
-        'At path: data.limitType -- Expected the value to satisfy a union of `literal | literal`, but received: "invalid"',
+        'At path: data.limitType -- Expected the value to satisfy a union of `literal | literal | literal | literal`, but received: "invalid"',
     },
     {
       name: 'invalid current type',

packages/kernel-errors/src/errors/ResourceLimitError.ts

@@ -29,7 +29,11 @@ export class ResourceLimitError extends BaseError {
     message: string,
     options?: ErrorOptionsWithStack & {
       data?: {
-        limitType?: 'connection' | 'messageSize';
+        limitType?:
+          | 'connection'
+          | 'messageSize'
+          | 'messageRate'
+          | 'connectionRate';
         current?: number;
         limit?: number;
       };
@@ -50,7 +54,12 @@ export class ResourceLimitError extends BaseError {
     data: optional(
       object({
         limitType: optional(
-          union([literal('connection'), literal('messageSize')]),
+          union([
+            literal('connection'),
+            literal('messageSize'),
+            literal('messageRate'),
+            literal('connectionRate'),
+          ]),
         ),
         current: optional(number()),
         limit: optional(number()),
@@ -75,7 +84,11 @@ export class ResourceLimitError extends BaseError {
     const options = unmarshalErrorOptions(marshaledError);
     const data = marshaledError.data as
       | {
-          limitType?: 'connection' | 'messageSize';
+          limitType?:
+            | 'connection'
+            | 'messageSize'
+            | 'messageRate'
+            | 'connectionRate';
           current?: number;
           limit?: number;
         }

packages/nodejs/test/e2e/remote-comms.test.ts

@@ -505,12 +505,14 @@ describe.sequential('Remote Communications E2E', () => {
     it(
       'rejects new messages when queue reaches MAX_QUEUE limit',
       async () => {
+        // Use high rate limit to avoid rate limiting interference with queue limit test
         const { aliceRef, bobURL } = await setupAliceAndBob(
           kernel1,
           kernel2,
           kernelStore1,
           kernelStore2,
           testRelays,
+          { maxMessagesPerSecond: 500 },
         );
 
         await sendRemoteMessage(kernel1, aliceRef, bobURL, 'hello', ['Alice']);

packages/nodejs/test/helpers/remote-comms.ts

@@ -1,7 +1,11 @@
 import type { KernelDatabase } from '@metamask/kernel-store';
 import { stringify } from '@metamask/kernel-utils';
 import { Kernel, kunser, makeKernelStore } from '@metamask/ocap-kernel';
-import type { ClusterConfig, KRef } from '@metamask/ocap-kernel';
+import type {
+  ClusterConfig,
+  KRef,
+  RemoteCommsOptions,
+} from '@metamask/ocap-kernel';
 
 import { makeTestKernel } from './kernel.ts';
 
@@ -181,6 +185,7 @@ export async function wait(ms: number): Promise<void> {
  * @param kernelStore1 - Kernel store for first kernel.
  * @param kernelStore2 - Kernel store for second kernel.
  * @param relays - Array of relay addresses.
+ * @param remoteCommsOptions - Optional additional options for initRemoteComms.
  * @returns Object with all setup data including URLs and references.
  */
 export async function setupAliceAndBob(
@@ -189,6 +194,7 @@ export async function setupAliceAndBob(
   kernelStore1: ReturnType<typeof makeKernelStore>,
   kernelStore2: ReturnType<typeof makeKernelStore>,
   relays: string[],
+  remoteCommsOptions?: Omit<RemoteCommsOptions, 'relays'>,
 ): Promise<{
   aliceURL: string;
   bobURL: string;
@@ -197,8 +203,8 @@ export async function setupAliceAndBob(
   peerId1: string;
   peerId2: string;
 }> {
-  await kernel1.initRemoteComms({ relays });
-  await kernel2.initRemoteComms({ relays });
+  await kernel1.initRemoteComms({ relays, ...remoteCommsOptions });
+  await kernel2.initRemoteComms({ relays, ...remoteCommsOptions });
 
   const aliceConfig = makeRemoteVatConfig('Alice');
   const bobConfig = makeRemoteVatConfig('Bob');

packages/ocap-kernel/src/remotes/platform/constants.ts

@@ -15,3 +15,15 @@ export const DEFAULT_WRITE_TIMEOUT_MS = 10_000;
 
 /** SCTP user initiated abort code (RFC 4960) */
 export const SCTP_USER_INITIATED_ABORT = 12;
+
+/** Default message rate limit: 100 messages per second per peer */
+export const DEFAULT_MESSAGE_RATE_LIMIT = 100;
+
+/** Default message rate window in milliseconds (1 second) */
+export const DEFAULT_MESSAGE_RATE_WINDOW_MS = 1000;
+
+/** Default connection attempt rate limit: 10 attempts per minute per peer */
+export const DEFAULT_CONNECTION_RATE_LIMIT = 10;
+
+/** Default connection rate window in milliseconds (1 minute) */
+export const DEFAULT_CONNECTION_RATE_WINDOW_MS = 60_000;