feat: moved rpcServerAgent into NodeConnectionManager

The agent level RPC is not fully managed by the Nodes domain.

[ci skip]

Author: tegefaulkes

src/PolykeyAgent.ts

@@ -36,7 +36,6 @@ import * as utils from './utils';
 import * as keysUtils from './keys/utils';
 import * as keysEvents from './keys/events';
 import * as nodesUtils from './nodes/utils';
-import * as nodesEvents from './nodes/events';
 import * as workersUtils from './workers/utils';
 import TaskManager from './tasks/TaskManager';
 import { serverManifest as clientServerManifest } from './client/handlers';
@@ -423,28 +422,6 @@ class PolykeyAgent {
         pingTimeoutTimeTime: optionsDefaulted.client.keepAliveTimeoutTime,
         logger: logger.getChild('WebSocketServer'),
       });
-      rpcServerAgent = await RPCServer.createRPCServer({
-        manifest: agentServerManifest({
-          acl: acl,
-          db: db,
-          keyRing: keyRing,
-          logger: logger,
-          nodeConnectionManager: nodeConnectionManager,
-          nodeGraph: nodeGraph,
-          nodeManager: nodeManager,
-          notificationsManager: notificationsManager,
-          sigchain: sigchain,
-          vaultManager: vaultManager,
-        }),
-        middlewareFactory: rpcUtilsMiddleware.defaultServerMiddlewareWrapper(
-          undefined,
-          optionsDefaulted.rpc.parserBufferSize,
-        ),
-        sensitive: true,
-        handlerTimeoutTime: optionsDefaulted.rpc.callTimeoutTime,
-        handlerTimeoutGraceTime: optionsDefaulted.rpc.callTimeoutTime + 2000,
-        logger: logger.getChild(RPCServer.name + 'Agent'),
-      });
     } catch (e) {
       logger.warn(`Failed Creating ${this.name}`);
       await rpcServerAgent?.destroy({ force:  true });
@@ -487,7 +464,6 @@ class PolykeyAgent {
       sessionManager,
       rpcServerClient,
       webSocketServerClient,
-      rpcServerAgent,
       fs,
       logger,
     });
@@ -534,11 +510,6 @@ class PolykeyAgent {
   public readonly rpcServerAgent: RPCServer;
   protected workerManager: PolykeyWorkerManagerInterface | undefined;
 
-  protected handleEventNodeStream = (e: nodesEvents.EventNodeConnectionStream) => {
-    const stream = e.detail;
-    this.rpcServerAgent.handleStream(stream);
-  };
-
   protected handleEventCertManagerCertChange = async (
     evt: keysEvents.EventCertManagerCertChange,
   ) => {
@@ -580,7 +551,6 @@ class PolykeyAgent {
     sessionManager,
     rpcServerClient,
     webSocketServerClient,
-    rpcServerAgent,
     fs,
     logger,
   }: {
@@ -604,7 +574,6 @@ class PolykeyAgent {
     sessionManager: SessionManager;
     rpcServerClient: RPCServer;
     webSocketServerClient: WebSocketServer;
-    rpcServerAgent: RPCServer;
     fs: FileSystem;
     logger: Logger;
   }) {
@@ -629,7 +598,6 @@ class PolykeyAgent {
     this.sessionManager = sessionManager;
     this.rpcServerClient = rpcServerClient;
     this.webSocketServerClient = webSocketServerClient;
-    this.rpcServerAgent = rpcServerAgent;
     this.fs = fs;
   }
 
@@ -713,14 +681,22 @@ class PolykeyAgent {
           this.rpcServerClient.handleStream(streamPair),
       });
       await this.nodeManager.start();
-      this.nodeConnectionManager.addEventListener(
-        nodesEvents.EventNodeConnectionStream.name,
-        this.handleEventNodeStream,
-      );
       await this.nodeConnectionManager.start({
         host: optionsDefaulted.agentServiceHost,
         port: optionsDefaulted.agentServicePort,
         ipv6Only: optionsDefaulted.ipv6Only,
+        manifest: agentServerManifest({
+          acl: this.acl,
+          db: this.db,
+          keyRing: this.keyRing,
+          logger: this.logger,
+          nodeConnectionManager: this.nodeConnectionManager,
+          nodeGraph: this.nodeGraph,
+          nodeManager: this.nodeManager,
+          notificationsManager: this.notificationsManager,
+          sigchain: this.sigchain,
+          vaultManager: this.vaultManager,
+        }),
       });
       await this.nodeGraph.start({ fresh });
       await this.nodeManager.syncNodeGraph(false);
@@ -764,10 +740,6 @@ class PolykeyAgent {
       await this.discovery?.stop();
       await this.nodeGraph?.stop();
       await this.nodeConnectionManager?.stop();
-      this.nodeConnectionManager.removeEventListener(
-        nodesEvents.EventNodeConnectionStream.name,
-        this.handleEventNodeStream,
-      );
       await this.nodeManager?.stop();
       await this.webSocketServerClient.stop(true);
       await this.identitiesManager?.stop();
@@ -804,10 +776,6 @@ class PolykeyAgent {
     await this.vaultManager.stop();
     await this.discovery.stop();
     await this.nodeConnectionManager.stop();
-    this.nodeConnectionManager.removeEventListener(
-      nodesEvents.EventNodeConnectionStream.name,
-      this.handleEventNodeStream,
-    );
     await this.nodeGraph.stop();
     await this.nodeManager.stop();
     await this.webSocketServerClient.stop(true);

src/nodes/NodeConnectionManager.ts

@@ -14,8 +14,8 @@ import type {
 } from './types';
 import type KeyRing from '../keys/KeyRing';
 import type { Key, CertificatePEM } from '../keys/types';
-import type { ConnectionData, Host, Hostname, Port } from '../network/types';
-import type { TLSConfig } from '../network/types';
+import type { ConnectionData, Host, Hostname, Port, TLSConfig } from '../network/types';
+import type { ServerManifest } from '../rpc/types';
 import type { HolePunchRelayMessage } from './agent/types';
 import Logger from '@matrixai/logger';
 import { withF } from '@matrixai/resources';
@@ -37,6 +37,8 @@ import manifestClientAgent from './agent/callers';
 import * as utils from '../utils';
 import config from '../config';
 import { running, status } from "@matrixai/async-init";
+import RPCServer from '../rpc/RPCServer';
+import * as rpcUtilsMiddleware from '../rpc/utils/middleware';
 
 type ManifestClientAgent = typeof manifestClientAgent;
 
@@ -109,6 +111,16 @@ class NodeConnectionManager {
    */
   public readonly connectionHolePunchIntervalTime: number;
 
+  /**
+   * Max parse buffer size before RPC parser throws an parse error.
+   */
+  public readonly rpcParserBufferSize: number;
+
+  /**
+   * default timeout for RPC handlers
+   */
+  public readonly rpcCallTimeoutTime: number;
+
   protected logger: Logger;
   protected keyRing: KeyRing;
   protected nodeGraph: NodeGraph;
@@ -134,6 +146,8 @@ class NodeConnectionManager {
 
   protected connectionLocks: LockBox<Lock> = new LockBox();
 
+  protected rpcServer?: RPCServer;
+
   /**
    * Dispatches a `EventNodeConnectionManagerClose` in response to any `NodeConnectionManager`
    * error event. Will trigger stop of the `NodeConnectionManager` via the
@@ -157,6 +171,11 @@ class NodeConnectionManager {
     }
   }
 
+  protected handleEventNodeConnectionStream = async (e: nodesEvents.EventNodeConnectionStream) => {
+    const stream = e.detail;
+    this.rpcServer!.handleStream(stream);
+  }
+
   /**
    * redispatches `QUICSOcket` or `QUICServer` error events as `NodeConnectionManager` error events.
    * This should trigger the destruction of the `NodeConnection` through the
@@ -245,6 +264,8 @@ class NodeConnectionManager {
         .clientKeepAliveIntervalTime,
       connectionHolePunchIntervalTime: config.defaultsSystem
         .nodesConnectionHolePunchIntervalTime,
+      rpcParserBufferSize: config.defaultsSystem.rpcParserBufferSize,
+      rpcCallTimeoutTime: config.defaultsSystem.rpcCallTimeoutTime,
     })
     this.logger = logger ?? new Logger(this.constructor.name);
     this.keyRing = keyRing;
@@ -261,6 +282,8 @@ class NodeConnectionManager {
     this.connectionKeepAliveTimeoutTime = optionsDefaulted.connectionKeepAliveTimeoutTime;
     this.connectionKeepAliveIntervalTime = optionsDefaulted.connectionKeepAliveIntervalTime;
     this.connectionHolePunchIntervalTime = optionsDefaulted.connectionHolePunchIntervalTime;
+    this.rpcParserBufferSize = optionsDefaulted.rpcParserBufferSize;
+    this.rpcCallTimeoutTime = optionsDefaulted.rpcCallTimeoutTime;
     // Note that all buffers allocated for crypto operations is using
     // `allocUnsafeSlow`. Which ensures that the underlying `ArrayBuffer`
     // is not shared. Also, all node buffers satisfy the `ArrayBuffer` interface.
@@ -343,18 +366,34 @@ class NodeConnectionManager {
     port = 0 as Port,
     reuseAddr = false,
     ipv6Only = false,
+    manifest = {},
   }: {
     host?: Host;
     port?: Port;
     reuseAddr?: boolean;
     ipv6Only?: boolean;
+    manifest?: ServerManifest;
   }) {
     const address = networkUtils.buildAddress(host, port);
     this.logger.info(`Start ${this.constructor.name} on ${address}`);
 
     // We should expect that seed nodes are already in the node manager
     // It should not be managed here!
 
+    // setting up RPCServer
+    this.rpcServer = await RPCServer.createRPCServer({
+      manifest,
+      middlewareFactory: rpcUtilsMiddleware.defaultServerMiddlewareWrapper(
+        undefined,
+        this.rpcParserBufferSize,
+      ),
+      sensitive: true,
+      handlerTimeoutTime: this.rpcCallTimeoutTime,
+      handlerTimeoutGraceTime: this.rpcCallTimeoutTime + 2000,
+      logger: this.logger.getChild(RPCServer.name),
+    });
+
+    // Setting up QUICSocket
     await this.quicSocket.start({
       host,
       port,
@@ -397,7 +436,6 @@ class NodeConnectionManager {
       EventAll.name,
       this.handleEventAll,
     );
-
     this.logger.info(`Started ${this.constructor.name}`);
   }
 
@@ -453,6 +491,7 @@ class NodeConnectionManager {
     await Promise.all(destroyProms);
     await this.quicServer.stop({ force: true });
     await this.quicSocket.stop({ force: true });
+    await this.rpcServer?.destroy({ force: true, reason: Error('TMP NCM stopping')}); // TODO
     this.logger.info(`Stopped ${this.constructor.name}`);
   }
 
@@ -928,6 +967,10 @@ class NodeConnectionManager {
     // Check if exists in map, this should never happen but better safe than sorry.
     if (this.connections.has(nodeIdString)) utils.never();
     // Setting up events
+    nodeConnection.addEventListener(
+      nodesEvents.EventNodeConnectionStream.name,
+      this.handleEventNodeConnectionStream,
+    )
     nodeConnection.addEventListener(
       EventAll.name,
       this.handleEventAll,
@@ -941,6 +984,10 @@ class NodeConnectionManager {
         // Already locked so already destroying
         if (this.connectionLocks.isLocked(nodeIdString)) return;
         await this.destroyConnection(nodeId);
+        nodeConnection.removeEventListener(
+          nodesEvents.EventNodeConnectionStream.name,
+          this.handleEventNodeConnectionStream,
+        )
         nodeConnection.removeEventListener(
           EventAll.name,
           this.handleEventAll,

src/nodes/types.ts

@@ -33,6 +33,8 @@ type NodesOptions = {
   connectionKeepAliveTimeoutTime: number;
   connectionKeepAliveIntervalTime: number;
   connectionHolePunchIntervalTime: number;
+  rpcParserBufferSize: number;
+  rpcCallTimeoutTime: number;
 };
 
 export type {

tests/nodes/NodeConnectionManager.lifecycle.test.ts

@@ -14,7 +14,6 @@ import NodeConnectionManager from '@/nodes/NodeConnectionManager';
 import { promise, sleep } from '@/utils';
 import * as nodesErrors from '@/nodes/errors';
 import NodeConnection from '@/nodes/NodeConnection';
-import RPCServer from '@/rpc/RPCServer';
 import * as tlsUtils from '../utils/tls';
 
 describe(`${NodeConnectionManager.name} lifecycle test`, () => {
@@ -35,7 +34,6 @@ describe(`${NodeConnectionManager.name} lifecycle test`, () => {
   let serverNodeIdEncoded: NodeIdEncoded;
   let keyRingPeer: KeyRing;
   let nodeConnectionManagerPeer: NodeConnectionManager;
-  let rpcServer: RPCServer;
   let serverAddress: NodeAddress;
 
   let keyRing: KeyRing;
@@ -76,13 +74,6 @@ describe(`${NodeConnectionManager.name} lifecycle test`, () => {
     await nodeConnectionManagerPeer.start({
       host: localHost,
     })
-    rpcServer = await RPCServer.createRPCServer({
-      handlerTimeoutGraceTime: 1000,
-      handlerTimeoutTime: 5000,
-      logger: logger.getChild(`${RPCServer.name}`),
-      manifest: {}, // TODO: test server manifest
-      sensitive: false,
-    });
 
     // Setting up client dependencies
     const keysPath = path.join(dataDir, 'keys');
@@ -121,7 +112,6 @@ describe(`${NodeConnectionManager.name} lifecycle test`, () => {
     await keyRing.stop();
     await keyRing.destroy();
 
-    await rpcServer.destroy({ force: true });
     await nodeConnectionManagerPeer.stop();
   });