feat: streams are now using a RPCStream interface that extends ReadableWritablePair and includes a cancel(reason?: any) method and generic metadata

- created `RPCStream` interface, updating code
- fixing tests
- client now includes metadata in remote errors
- `RPCServer` includes metadata in the `RPCStream` and propagates cancel/meta to middleware
- improved cancellation feature of the server `RPCStream`

[ci skip]

Author: tegefaulkes

src/PolykeyAgent.ts

@@ -452,7 +452,7 @@ class PolykeyAgent {
         webSocketServerClient ??
         (await WebSocketServer.createWebSocketServer({
           connectionCallback: (streamPair) =>
-            rpcServerClient!.handleStream(streamPair, {}),
+            rpcServerClient!.handleStream(streamPair),
           fs,
           host: networkConfig_.clientHost,
           port: networkConfig_.clientPort,
@@ -775,7 +775,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/bin/utils/utils.ts

@@ -112,10 +112,7 @@ function outputFormatter(msg: OutputObject): string | Uint8Array {
     let currError = msg.data;
     let indent = '  ';
     while (currError != null) {
-      if (
-        currError instanceof grpcErrors.ErrorPolykeyRemoteOLD ||
-        currError instanceof rpcErrors.ErrorPolykeyRemote
-      ) {
+      if (currError instanceof grpcErrors.ErrorPolykeyRemoteOLD) {
         output += `${currError.name}: ${currError.description}`;
         if (currError.message && currError.message !== '') {
           output += ` - ${currError.message}`;
@@ -130,6 +127,20 @@ function outputFormatter(msg: OutputObject): string | Uint8Array {
         output += `${indent}timestamp\t${currError.timestamp}\n`;
         output += `${indent}cause: `;
         currError = currError.cause;
+      } else if (currError instanceof rpcErrors.ErrorPolykeyRemote) {
+        output += `${currError.name}: ${currError.description}`;
+        if (currError.message && currError.message !== '') {
+          output += ` - ${currError.message}`;
+        }
+        if (currError.metadata != null) {
+          output += '\n';
+          for (const [key, value] of Object.entries(currError.metadata)) {
+            output += `${indent}${key}\t${value}\n`;
+          }
+        }
+        output += `${indent}timestamp\t${currError.timestamp}\n`;
+        output += `${indent}cause: `;
+        currError = currError.cause;
       } else if (currError instanceof ErrorPolykey) {
         output += `${currError.name}: ${currError.description}`;
         if (currError.message && currError.message !== '') {

src/client/handlers/gestaltsGestaltList.ts

@@ -15,7 +15,8 @@ class GestaltsGestaltListHandler extends ServerHandler<
 > {
   public async *handle(
     _input,
-    _connectionInfo,
+    _cancel,
+    _meta,
     ctx,
   ): AsyncGenerator<ClientRPCResponseResult<GestaltMessage>> {
     const { db, gestaltGraph } = this.container;

src/client/handlers/identitiesAuthenticate.ts

@@ -21,7 +21,8 @@ class IdentitiesAuthenticateHandler extends ServerHandler<
     input: ClientRPCRequestParams<{
       providerId: string;
     }>,
-    _,
+    _cancel,
+    _meta,
     ctx,
   ): AsyncGenerator<ClientRPCResponseResult<AuthProcessMessage>> {
     if (ctx.signal.aborted) throw ctx.signal.reason;

src/client/handlers/identitiesAuthenticatedGet.ts

@@ -20,7 +20,8 @@ class IdentitiesAuthenticatedGetHandler extends ServerHandler<
     input: ClientRPCRequestParams<{
       providerId?: string;
     }>,
-    _,
+    _cancel,
+    _meta,
     ctx,
   ): AsyncGenerator<ClientRPCResponseResult<IdentityMessage>> {
     if (ctx.signal.aborted) throw ctx.signal.reason;

src/client/handlers/identitiesInfoConnectedGet.ts

@@ -18,7 +18,8 @@ class IdentitiesInfoConnectedGetHandler extends ServerHandler<
 > {
   public async *handle(
     input: ClientRPCRequestParams<ProviderSearchMessage>,
-    _,
+    _cancel,
+    _meta,
     ctx,
   ): AsyncGenerator<ClientRPCResponseResult<IdentityInfoMessage>> {
     if (ctx.signal.aborted) throw ctx.signal.reason;

src/client/handlers/identitiesInfoGet.ts

@@ -19,7 +19,8 @@ class IdentitiesInfoGetHandler extends ServerHandler<
 > {
   public async *handle(
     input: ClientRPCRequestParams<ProviderSearchMessage>,
-    _,
+    _cancel,
+    _meta,
     ctx,
   ): AsyncGenerator<ClientRPCResponseResult<IdentityInfoMessage>> {
     if (ctx.signal.aborted) throw ctx.signal.reason;

src/client/handlers/keysCertsChainGet.ts

@@ -12,7 +12,8 @@ class KeysCertsChainGetHandler extends ServerHandler<
 > {
   public async *handle(
     _input,
-    _connectionInfo,
+    _cancel,
+    _meta,
     ctx,
   ): AsyncGenerator<ClientRPCResponseResult<CertMessage>> {
     const { certManager } = this.container;

src/client/handlers/nodesGetAll.ts

@@ -17,7 +17,8 @@ class NodesGetAllHandler extends ServerHandler<
 > {
   public async *handle(
     _input,
-    _connectionUInfo,
+    _cancel,
+    _meta,
     ctx,
   ): AsyncGenerator<ClientRPCResponseResult<NodesGetMessage>> {
     if (ctx.signal.aborted) throw ctx.signal.reason;

src/client/handlers/nodesListConnections.ts

@@ -13,7 +13,8 @@ class NodesListConnectionsHandler extends ServerHandler<
 > {
   public async *handle(
     _input,
-    _connectionInfo,
+    _cancel,
+    _meta,
     ctx,
   ): AsyncGenerator<ClientRPCResponseResult<NodeConnectionMessage>> {
     const { nodeConnectionManager } = this.container;