Merge pull request #513 from MatrixAI/feature-agnostic_timeout

RPC timeout implementation

Author: tegefaulkes

package-lock.json

@@ -80,15 +80,15 @@
   },
   "dependencies": {
     "@grpc/grpc-js": "1.6.7",
-    "@matrixai/async-cancellable": "^1.0.2",
+    "@matrixai/async-cancellable": "^1.0.4",
     "@matrixai/async-init": "^1.8.2",
     "@matrixai/async-locks": "^3.2.0",
     "@matrixai/db": "^5.1.0",
     "@matrixai/errors": "^1.1.7",
     "@matrixai/id": "^3.3.3",
     "@matrixai/logger": "^3.1.0",
     "@matrixai/resources": "^1.1.4",
-    "@matrixai/timer": "^1.0.0",
+    "@matrixai/timer": "^1.1.0",
     "@matrixai/workers": "^1.3.6",
     "@peculiar/asn1-pkcs8": "^2.3.0",
     "@peculiar/asn1-schema": "^2.3.0",
@@ -128,7 +128,7 @@
   "devDependencies": {
     "@babel/preset-env": "^7.13.10",
     "@fast-check/jest": "^1.1.0",
-    "@streamparser/json": "^0.0.12",
+    "@streamparser/json": "^0.0.13",
     "@swc/core": "^1.2.215",
     "@types/cross-spawn": "^6.0.2",
     "@types/google-protobuf": "^3.7.4",

package.json

@@ -12,8 +12,8 @@ import { DB } from '@matrixai/db';
 import { CreateDestroyStartStop } from '@matrixai/async-init/dist/CreateDestroyStartStop';
 import RPCServer from './rpc/RPCServer';
 import WebSocketServer from './websockets/WebSocketServer';
-import * as middlewareUtils from './rpc/utils/middleware';
-import * as authMiddleware from './client/utils/authenticationMiddleware';
+import * as rpcUtilsMiddleware from './rpc/utils/middleware';
+import * as clientUtilsMiddleware from './client/utils/middleware';
 import { WorkerManager } from './workers';
 import * as networkUtils from './network/utils';
 import KeyRing from './keys/KeyRing';
@@ -56,10 +56,13 @@ type NetworkConfig = {
   // RPCServer for client service
   clientHost?: Host;
   clientPort?: Port;
-  maxReadBufferBytes?: number;
-  idleTimeout?: number;
-  pingInterval?: number;
-  pingTimeout?: number;
+  maxReadableStreamBytes?: number;
+  connectionIdleTimeoutTime?: number;
+  pingIntervalTime?: number;
+  pingTimeoutTime?: number;
+  handlerTimeoutTime?: number;
+  handlerTimeoutGraceTime?: number;
+  clientParserBufferByteLimit?: number;
 };
 
 interface PolykeyAgent extends CreateDestroyStartStop {}
@@ -193,7 +196,7 @@ class PolykeyAgent {
       ...config.defaults.nodeConnectionManagerConfig,
       ...utils.filterEmptyObject(nodeConnectionManagerConfig),
     };
-    const _networkConfig = {
+    const networkConfig_ = {
       ...config.defaults.networkConfig,
       ...utils.filterEmptyObject(networkConfig),
     };
@@ -437,13 +440,13 @@ class PolykeyAgent {
             sessionManager: sessionManager,
             vaultManager: vaultManager,
           }),
-          middlewareFactory: middlewareUtils.defaultServerMiddlewareWrapper(
-            authMiddleware.authenticationMiddlewareServer(
-              sessionManager,
-              keyRing,
-            ),
+          middlewareFactory: rpcUtilsMiddleware.defaultServerMiddlewareWrapper(
+            clientUtilsMiddleware.middlewareServer(sessionManager, keyRing),
+            networkConfig_.clientParserBufferByteLimit,
           ),
           sensitive: false,
+          handlerTimeoutTime: networkConfig_.handlerTimeoutTime,
+          handlerTimeoutGraceTime: networkConfig_.handlerTimeoutGraceTime,
           logger: logger.getChild('RPCServerClient'),
         });
       }
@@ -454,16 +457,16 @@ class PolykeyAgent {
       webSocketServerClient =
         webSocketServerClient ??
         (await WebSocketServer.createWebSocketServer({
-          connectionCallback: (streamPair) =>
-            rpcServerClient!.handleStream(streamPair, {}),
+          connectionCallback: (rpcStream) =>
+            rpcServerClient!.handleStream(rpcStream),
           fs,
-          host: _networkConfig.clientHost,
-          port: _networkConfig.clientPort,
+          host: networkConfig_.clientHost,
+          port: networkConfig_.clientPort,
           tlsConfig,
-          maxReadBufferBytes: _networkConfig.maxReadBufferBytes,
-          idleTimeout: _networkConfig.idleTimeout,
-          pingInterval: _networkConfig.pingInterval,
-          pingTimeout: _networkConfig.pingTimeout,
+          maxReadableStreamBytes: networkConfig_.maxReadableStreamBytes,
+          connectionIdleTimeoutTime: networkConfig_.connectionIdleTimeoutTime,
+          pingIntervalTime: networkConfig_.pingIntervalTime,
+          pingTimeoutTime: networkConfig_.pingTimeoutTime,
           logger: logger.getChild('WebSocketServer'),
         }));
       grpcServerAgent =
@@ -778,7 +781,7 @@ class PolykeyAgent {
         host: _networkConfig.clientHost,
         port: _networkConfig.clientPort,
         connectionCallback: (streamPair) =>
-          this.rpcServerClient.handleStream(streamPair, {}),
+          this.rpcServerClient.handleStream(streamPair),
       });
       // Agent server
       await this.grpcServerAgent.start({

src/PolykeyAgent.ts

@@ -4,8 +4,8 @@ import path from 'path';
 import Logger from '@matrixai/logger';
 import { CreateDestroyStartStop } from '@matrixai/async-init/dist/CreateDestroyStartStop';
 import RPCClient from './rpc/RPCClient';
-import * as middlewareUtils from './rpc/utils/middleware';
-import * as authMiddleware from './client/utils/authenticationMiddleware';
+import * as rpcUtilsMiddleware from './rpc/utils/middleware';
+import * as clientUtilsMiddleware from './client/utils/middleware';
 import { Session } from './sessions';
 import * as errors from './errors';
 import * as utils from './utils';
@@ -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;
@@ -60,9 +64,12 @@ class PolykeyClient<M extends ClientManifest> {
         : await RPCClient.createRPCClient<M>({
             manifest,
             streamFactory,
-            middlewareFactory: middlewareUtils.defaultClientMiddlewareWrapper(
-              authMiddleware.authenticationMiddlewareClient(session),
-            ),
+            middlewareFactory:
+              rpcUtilsMiddleware.defaultClientMiddlewareWrapper(
+                clientUtilsMiddleware.middlewareClient(session),
+                parserBufferByteLimit,
+              ),
+            streamKeepAliveTimeoutTime,
             logger: logger.getChild(RPCClient.name),
           });
     const pkClient = new this({

src/PolykeyClient.ts

@@ -58,7 +58,7 @@ class CommandLockAll extends CommandPolykey {
           logger: this.logger.getChild(WebSocketClient.name),
         });
         pkClient = await PolykeyClient.createPolykeyClient({
-          streamFactory: () => webSocketClient.startConnection(),
+          streamFactory: (ctx) => webSocketClient.startConnection(ctx),
           nodePath: options.nodePath,
           manifest: clientManifest,
           logger: this.logger.getChild(PolykeyClient.name),

src/bin/agent/CommandLockAll.ts

@@ -63,7 +63,7 @@ class CommandStatus extends CommandPolykey {
             logger: this.logger.getChild(WebSocketClient.name),
           });
           pkClient = await PolykeyClient.createPolykeyClient({
-            streamFactory: () => webSocketClient.startConnection(),
+            streamFactory: (ctx) => webSocketClient.startConnection(ctx),
             nodePath: options.nodePath,
             manifest: clientManifest,
             logger: this.logger.getChild(PolykeyClient.name),

src/bin/agent/CommandStatus.ts

@@ -60,7 +60,7 @@ class CommandStop extends CommandPolykey {
           logger: this.logger.getChild(WebSocketClient.name),
         });
         pkClient = await PolykeyClient.createPolykeyClient({
-          streamFactory: () => webSocketClient.startConnection(),
+          streamFactory: (ctx) => webSocketClient.startConnection(ctx),
           nodePath: options.nodePath,
           manifest: clientManifest,
           logger: this.logger.getChild(PolykeyClient.name),

src/bin/agent/CommandStop.ts

@@ -47,7 +47,7 @@ class CommandUnlock extends CommandPolykey {
           logger: this.logger.getChild(WebSocketClient.name),
         });
         pkClient = await PolykeyClient.createPolykeyClient({
-          streamFactory: () => webSocketClient.startConnection(),
+          streamFactory: (ctx) => webSocketClient.startConnection(ctx),
           nodePath: options.nodePath,
           manifest: clientManifest,
           logger: this.logger.getChild(PolykeyClient.name),

src/bin/agent/CommandUnlock.ts

@@ -61,7 +61,7 @@ class CommandAllow extends CommandPolykey {
           logger: this.logger.getChild(WebSocketClient.name),
         });
         pkClient = await PolykeyClient.createPolykeyClient({
-          streamFactory: () => webSocketClient.startConnection(),
+          streamFactory: (ctx) => webSocketClient.startConnection(ctx),
           nodePath: options.nodePath,
           manifest: clientManifest,
           logger: this.logger.getChild(PolykeyClient.name),

src/bin/identities/CommandAllow.ts

@@ -58,7 +58,7 @@ class CommandAuthenticate extends CommandPolykey {
           logger: this.logger.getChild(WebSocketClient.name),
         });
         pkClient = await PolykeyClient.createPolykeyClient({
-          streamFactory: () => webSocketClient.startConnection(),
+          streamFactory: (ctx) => webSocketClient.startConnection(ctx),
           nodePath: options.nodePath,
           manifest: clientManifest,
           logger: this.logger.getChild(PolykeyClient.name),