fix: adding and propagating config values

[ci skip]

Author: tegefaulkes

src/PolykeyAgent.ts

@@ -56,10 +56,13 @@ type NetworkConfig = {
   // RPCServer for client service
   clientHost?: Host;
   clientPort?: Port;
-  maxReadBufferBytes?: number;
-  idleTimeoutTime?: number;
+  maxReadableStreamBytes?: number;
+  connectionIdleTimeoutTime?: number;
   pingIntervalTime?: number;
   pingTimeoutTime?: number;
+  handlerTimeoutTime?: number;
+  handlerTimeoutGraceTime?: number;
+  clientParserBufferByteLimit?: number;
 };
 
 interface PolykeyAgent extends CreateDestroyStartStop {}
@@ -439,8 +442,11 @@ class PolykeyAgent {
           }),
           middlewareFactory: rpcUtilsMiddleware.defaultServerMiddlewareWrapper(
             clientUtilsMiddleware.middlewareServer(sessionManager, keyRing),
+            networkConfig_.clientParserBufferByteLimit,
           ),
           sensitive: false,
+          handlerTimeoutTime: networkConfig_.handlerTimeoutTime,
+          handlerTimeoutGraceTime: networkConfig_.handlerTimeoutGraceTime,
           logger: logger.getChild('RPCServerClient'),
         });
       }
@@ -451,14 +457,14 @@ class PolykeyAgent {
       webSocketServerClient =
         webSocketServerClient ??
         (await WebSocketServer.createWebSocketServer({
-          connectionCallback: (streamPair) =>
-            rpcServerClient!.handleStream(streamPair),
+          connectionCallback: (rpcStream) =>
+            rpcServerClient!.handleStream(rpcStream),
           fs,
           host: networkConfig_.clientHost,
           port: networkConfig_.clientPort,
           tlsConfig,
-          maxReadableStreamBytes: networkConfig_.maxReadBufferBytes,
-          connectionIdleTimeoutTime: networkConfig_.idleTimeout,
+          maxReadableStreamBytes: networkConfig_.maxReadableStreamBytes,
+          connectionIdleTimeoutTime: networkConfig_.connectionIdleTimeoutTime,
           pingIntervalTime: networkConfig_.pingIntervalTime,
           pingTimeoutTime: networkConfig_.pingTimeoutTime,
           logger: logger.getChild('WebSocketServer'),

src/PolykeyClient.ts

@@ -29,6 +29,8 @@ class PolykeyClient<M extends ClientManifest> {
     session,
     manifest,
     streamFactory,
+    streamKeepAliveTimeoutTime,
+    parserBufferByteLimit,
     fs = require('fs'),
     logger = new Logger(this.name),
     fresh = false,
@@ -37,6 +39,8 @@ class PolykeyClient<M extends ClientManifest> {
     session?: Session;
     manifest: RPCClient<M> | M;
     streamFactory: StreamFactory;
+    streamKeepAliveTimeoutTime?: number;
+    parserBufferByteLimit?: number;
     fs?: FileSystem;
     logger?: Logger;
     fresh?: boolean;
@@ -63,7 +67,9 @@ class PolykeyClient<M extends ClientManifest> {
             middlewareFactory:
               rpcUtilsMiddleware.defaultClientMiddlewareWrapper(
                 clientUtilsMiddleware.middlewareClient(session),
+                parserBufferByteLimit,
               ),
+            streamKeepAliveTimeoutTime,
             logger: logger.getChild(RPCClient.name),
           });
     const pkClient = new this({

src/config.ts

@@ -98,12 +98,17 @@ const config = {
       // GRPCServer for agent service
       agentHost: '127.0.0.1' as Host,
       agentPort: 0 as Port,
-      // GRPCServer for client service
+      // RPCServer for client service
       clientHost: '127.0.0.1' as Host,
       clientPort: 0 as Port,
-      maxReadBufferBytes: 1_000_000_000, // About 1 GB
+      // Times and limits for client communication
+      connectionIdleTimeoutTime: 120, // 2 minutes
       pingIntervalTime: 1_000, // 1 second
       pingTimeoutTime: 10_000, // 10 seconds
+      handlerTimeoutTime: 60_000, // 1 minute
+      handlerTimeoutGraceTime: 2_000, // 2 seconds
+      maxReadableStreamBytes: 1_000_000_000, // About 1 GB
+      clientParserBufferByteLimit: 1_000_000, // About 1MB
     },
     proxyConfig: {
       connConnectTime: 2000,

src/rpc/RPCServer.ts

@@ -70,8 +70,8 @@ class RPCServer extends EventTarget {
     manifest,
     middlewareFactory = rpcUtilsMiddleware.defaultServerMiddlewareWrapper(),
     sensitive = false,
-    streamKeepAliveTimeoutTime = 60_000, // 1 minute
-    timeoutForceCloseTime = 2_000, // 2 seconds
+    handlerTimeoutTime = 60_000, // 1 minute
+    handlerTimeoutGraceTime = 2_000, // 2 seconds
     logger = new Logger(this.name),
   }: {
     manifest: ServerManifest;
@@ -82,17 +82,17 @@ class RPCServer extends EventTarget {
       JSONRPCResponse
     >;
     sensitive?: boolean;
-    streamKeepAliveTimeoutTime?: number;
-    timeoutForceCloseTime?: number;
+    handlerTimeoutTime?: number;
+    handlerTimeoutGraceTime?: number;
     logger?: Logger;
   }): Promise<RPCServer> {
     logger.info(`Creating ${this.name}`);
     const rpcServer = new this({
       manifest,
       middlewareFactory,
       sensitive,
-      streamKeepAliveTimeoutTime: streamKeepAliveTimeoutTime,
-      timeoutForceCloseTime: timeoutForceCloseTime,
+      handlerTimeoutTime,
+      handlerTimeoutGraceTime,
       logger,
     });
     logger.info(`Created ${this.name}`);
@@ -102,8 +102,8 @@ class RPCServer extends EventTarget {
   protected logger: Logger;
   protected handlerMap: Map<string, RawHandlerImplementation> = new Map();
   protected defaultTimeoutMap: Map<string, number | undefined> = new Map();
-  protected streamKeepAliveTimeoutTime: number;
-  protected timeoutForceCloseTime: number;
+  protected handlerTimeoutTime: number;
+  protected handlerTimeoutGraceTime: number;
   protected activeStreams: Set<PromiseCancellable<void>> = new Set();
   protected sensitive: boolean;
   protected middlewareFactory: MiddlewareFactory<
@@ -117,8 +117,8 @@ class RPCServer extends EventTarget {
     manifest,
     middlewareFactory,
     sensitive,
-    streamKeepAliveTimeoutTime = 60_000, // 1 minuet
-    timeoutForceCloseTime = 2_000, // 2 seconds
+    handlerTimeoutTime = 60_000, // 1 minuet
+    handlerTimeoutGraceTime = 2_000, // 2 seconds
     logger,
   }: {
     manifest: ServerManifest;
@@ -129,8 +129,8 @@ class RPCServer extends EventTarget {
       Uint8Array,
       JSONRPCResponseResult
     >;
-    streamKeepAliveTimeoutTime?: number;
-    timeoutForceCloseTime?: number;
+    handlerTimeoutTime?: number;
+    handlerTimeoutGraceTime?: number;
     sensitive: boolean;
     logger: Logger;
   }) {
@@ -188,8 +188,8 @@ class RPCServer extends EventTarget {
     }
     this.middlewareFactory = middlewareFactory;
     this.sensitive = sensitive;
-    this.streamKeepAliveTimeoutTime = streamKeepAliveTimeoutTime;
-    this.timeoutForceCloseTime = timeoutForceCloseTime;
+    this.handlerTimeoutTime = handlerTimeoutTime;
+    this.handlerTimeoutGraceTime = handlerTimeoutGraceTime;
     this.logger = logger;
   }
 
@@ -413,7 +413,7 @@ class RPCServer extends EventTarget {
     const abortController = new AbortController();
     // Setting up timeout timer logic
     const timer = new Timer({
-      delay: this.streamKeepAliveTimeoutTime,
+      delay: this.handlerTimeoutTime,
       handler: () => {
         abortController.abort(new rpcErrors.ErrorRPCTimedOut());
       },
@@ -424,7 +424,7 @@ class RPCServer extends EventTarget {
     let graceTimer: Timer<void> | undefined;
     const handleAbort = () => {
       const graceTimer = new Timer({
-        delay: this.timeoutForceCloseTime,
+        delay: this.handlerTimeoutGraceTime,
         handler: () => {
           rpcStream.cancel(abortController.signal.reason);
         },
@@ -501,7 +501,7 @@ class RPCServer extends EventTarget {
       }
       // Setting up Timeout logic
       const timeout = this.defaultTimeoutMap.get(method);
-      if (timeout != null && timeout < this.streamKeepAliveTimeoutTime) {
+      if (timeout != null && timeout < this.handlerTimeoutTime) {
         // Reset timeout with new delay if it is less than the default
         timer.reset(timeout);
       } else {

src/rpc/utils/middleware.ts

@@ -18,12 +18,12 @@ import { promise } from '../../utils/index';
  * also infers the type of the stream output.
  * @param messageParser - Validates the JSONRPC messages, so you can select for a
  *  specific type of message
- * @param byteLimit - sets the number of bytes buffered before throwing an
+ * @param bufferByteLimit - sets the number of bytes buffered before throwing an
  *  error. This is used to avoid infinitely buffering the input.
  */
 function binaryToJsonMessageStream<T extends JSONRPCMessage>(
   messageParser: (message: unknown) => T,
-  byteLimit: number = 1024 * 1024,
+  bufferByteLimit: number = 1024 * 1024,
 ): TransformStream<Uint8Array, T> {
   const parser = new JSONParser({
     separator: '',
@@ -53,7 +53,7 @@ function binaryToJsonMessageStream<T extends JSONRPCMessage>(
       } catch (e) {
         throw new rpcErrors.ErrorRPCParse(undefined, { cause: e });
       }
-      if (bytesWritten > byteLimit) {
+      if (bytesWritten > bufferByteLimit) {
         throw new rpcErrors.ErrorRPCMessageLength();
       }
     },
@@ -95,6 +95,7 @@ function defaultMiddleware() {
  * The reverse path will pipe the output stream through the provided middleware
  * and then transform it back to a binary stream.
  * @param middlewareFactory - The provided middleware
+ * @param parserBufferByteLimit
  */
 function defaultServerMiddlewareWrapper(
   middlewareFactory: MiddlewareFactory<
@@ -103,10 +104,12 @@ function defaultServerMiddlewareWrapper(
     JSONRPCResponse,
     JSONRPCResponse
   > = defaultMiddleware,
+  parserBufferByteLimit: number = 1024 * 1024,
 ): MiddlewareFactory<JSONRPCRequest, Uint8Array, Uint8Array, JSONRPCResponse> {
   return (ctx, cancel, meta) => {
     const inputTransformStream = binaryToJsonMessageStream(
       rpcUtils.parseJSONRPCRequest,
+      parserBufferByteLimit,
     );
     const outputTransformStream = new TransformStream<
       JSONRPCResponseResult,
@@ -143,6 +146,8 @@ function defaultServerMiddlewareWrapper(
  * The reverse path will parse and validate the output and pipe it through the
  * provided middleware.
  * @param middleware - the provided middleware
+ * @param parserBufferByteLimit - Max number of bytes to buffer when parsing the stream. Exceeding this results in an
+ * `ErrorRPCMessageLength` error.
  */
 const defaultClientMiddlewareWrapper = (
   middleware: MiddlewareFactory<
@@ -151,6 +156,7 @@ const defaultClientMiddlewareWrapper = (
     JSONRPCResponse,
     JSONRPCResponse
   > = defaultMiddleware,
+  parserBufferByteLimit?: number,
 ): MiddlewareFactory<
   Uint8Array,
   JSONRPCRequest,
@@ -160,7 +166,7 @@ const defaultClientMiddlewareWrapper = (
   return (ctx, cancel, meta) => {
     const outputTransformStream = binaryToJsonMessageStream(
       rpcUtils.parseJSONRPCResponse,
-      // Undefined,
+      parserBufferByteLimit,
     );
     const inputTransformStream = new TransformStream<
       JSONRPCRequest,

tests/client/timeoutMiddleware.test.ts

@@ -164,7 +164,7 @@ describe('timeoutMiddleware', () => {
       middlewareFactory: rpcUtilsMiddleware.defaultServerMiddlewareWrapper(
         timeoutMiddleware.timeoutMiddlewareServer,
       ),
-      streamKeepAliveTimeoutTime: 100,
+      handlerTimeoutTime: 100,
       logger,
     });
     clientServer = await WebSocketServer.createWebSocketServer({

tests/rpc/RPCServer.test.ts

@@ -789,7 +789,7 @@ describe(`${RPCServer.name}`, () => {
       manifest: {
         testMethod: new TestHandler({}),
       },
-      streamKeepAliveTimeoutTime: 100,
+      handlerTimeoutTime: 100,
       logger,
     });
     const [outputResult, outputStream] = rpcTestUtils.streamToArray();
@@ -821,7 +821,7 @@ describe(`${RPCServer.name}`, () => {
   test('timeout with default time before handler selected', async () => {
     const rpcServer = await RPCServer.createRPCServer({
       manifest: {},
-      streamKeepAliveTimeoutTime: 100,
+      handlerTimeoutTime: 100,
       logger,
     });
     const readWriteStream: RPCStream<Uint8Array, Uint8Array> = {
@@ -880,7 +880,7 @@ describe(`${RPCServer.name}`, () => {
           testShort: new TestMethodShortTimeout({}),
           testLong: new TestMethodLongTimeout({}),
         },
-        streamKeepAliveTimeoutTime: 50,
+        handlerTimeoutTime: 50,
         logger,
       });
       const streamShort = rpcTestUtils.messagesToReadableStream([
@@ -1045,8 +1045,8 @@ describe(`${RPCServer.name}`, () => {
       manifest: {
         testMethod: new TestHandler({}),
       },
-      streamKeepAliveTimeoutTime: 50,
-      timeoutForceCloseTime: 100,
+      handlerTimeoutTime: 50,
+      handlerTimeoutGraceTime: 100,
       logger,
     });
     const [, outputStream] = rpcTestUtils.streamToArray();
@@ -1136,7 +1136,7 @@ describe(`${RPCServer.name}`, () => {
       manifest: {
         testMethod: new TestHandler({}),
       },
-      timeoutForceCloseTime: 0,
+      handlerTimeoutGraceTime: 0,
       logger,
     });
     const [, outputStream] = rpcTestUtils.streamToArray();