Merge pull request #475 from MatrixAI/feature-nodes_cancellability

Updating the `Nodes` domain with `timedCancellable`
MatrixAI · Oct 7, 2022 · e428206 · e428206
2 parents 06df14a + 69c0ffe
commit e428206
Show file tree

Hide file tree

Showing 16 changed files with 638 additions and 411 deletions.
diff --git a/src/PolykeyAgent.ts b/src/PolykeyAgent.ts
@@ -672,7 +672,7 @@ class PolykeyAgent {
       await this.nodeManager.start();
       await this.nodeConnectionManager.start({ nodeManager: this.nodeManager });
       await this.nodeGraph.start({ fresh });
-      await this.nodeManager.syncNodeGraph(false);
+      await this.nodeManager.syncNodeGraph(false, 2000);
       await this.discovery.start({ fresh });
       await this.vaultManager.start({ fresh });
       await this.notificationsManager.start({ fresh });

diff --git a/src/contexts/decorators/cancellable.ts b/src/contexts/decorators/cancellable.ts
@@ -25,8 +25,12 @@ function cancellable(lazy: boolean = false) {
       let ctx: Partial<ContextCancellable> = args[contextIndex];
       if (ctx === undefined) {
         ctx = {};
-        args[contextIndex] = ctx;
+      } else {
+        // Copy the ctx into a new ctx object to avoid mutating the ctx in case
+        // it is used again
+        ctx = { ...ctx };
       }
+      args[contextIndex] = ctx;
       // Runtime type check on the context parameter
       contextsUtils.checkContextCancellable(ctx, key, targetName);
       return setupCancellable(

diff --git a/src/contexts/decorators/timed.ts b/src/contexts/decorators/timed.ts
@@ -31,8 +31,12 @@ function timed(
         let ctx: Partial<ContextTimed> = args[contextIndex];
         if (ctx === undefined) {
           ctx = {};
-          args[contextIndex] = ctx;
+        } else {
+          // Copy the ctx into a new ctx object to avoid mutating the ctx in case
+          // it is used again
+          ctx = { ...ctx };
         }
+        args[contextIndex] = ctx;
         // Runtime type check on the context parameter
         contextsUtils.checkContextTimed(ctx, key, targetName);
         const teardownContext = setupTimedContext(
@@ -51,8 +55,12 @@ function timed(
         let ctx: Partial<ContextTimed> = args[contextIndex];
         if (ctx === undefined) {
           ctx = {};
-          args[contextIndex] = ctx;
+        } else {
+          // Copy the ctx into a new ctx object to avoid mutating the ctx in case
+          // it is used again
+          ctx = { ...ctx };
         }
+        args[contextIndex] = ctx;
         // Runtime type check on the context parameter
         contextsUtils.checkContextTimed(ctx, key, targetName);
         const teardownContext = setupTimedContext(
@@ -71,8 +79,12 @@ function timed(
         let ctx: Partial<ContextTimed> = args[contextIndex];
         if (ctx === undefined) {
           ctx = {};
-          args[contextIndex] = ctx;
+        } else {
+          // Copy the ctx into a new ctx object to avoid mutating the ctx in case
+          // it is used again
+          ctx = { ...ctx };
         }
+        args[contextIndex] = ctx;
         // Runtime type check on the context parameter
         contextsUtils.checkContextTimed(ctx, key, targetName);
         const teardownContext = setupTimedContext(
@@ -91,8 +103,12 @@ function timed(
         let ctx: Partial<ContextTimed> = args[contextIndex];
         if (ctx === undefined) {
           ctx = {};
-          args[contextIndex] = ctx;
+        } else {
+          // Copy the ctx into a new ctx object to avoid mutating the ctx in case
+          // it is used again
+          ctx = { ...ctx };
         }
+        args[contextIndex] = ctx;
         // Runtime type check on the context parameter
         contextsUtils.checkContextTimed(ctx, key, targetName);
         const teardownContext = setupTimedContext(

diff --git a/src/contexts/decorators/timedCancellable.ts b/src/contexts/decorators/timedCancellable.ts
@@ -31,8 +31,12 @@ function timedCancellable(
       let ctx: Partial<ContextTimed> = args[contextIndex];
       if (ctx === undefined) {
         ctx = {};
-        args[contextIndex] = ctx;
+      } else {
+        // Copy the ctx into a new ctx object to avoid mutating the ctx in case
+        // it is used again
+        ctx = { ...ctx };
       }
+      args[contextIndex] = ctx;
       // Runtime type check on the context parameter
       contextsUtils.checkContextTimed(ctx, key, targetName);
       const lazy_ = typeof lazy === 'boolean' ? lazy : lazy(this);

diff --git a/src/contexts/functions/cancellable.ts b/src/contexts/functions/cancellable.ts
@@ -73,7 +73,7 @@ function cancellable<C extends ContextCancellable, P extends Array<any>, R>(
   lazy: boolean = false,
 ): (...params: ContextAndParameters<C, P>) => PromiseCancellable<R> {
   return (...params) => {
-    const ctx = params[0] ?? {};
+    const ctx = params[0] != null ? { ...params[0] } : {};
     const args = params.slice(1) as P;
     return setupCancellable(f, lazy, ctx, args);
   };

diff --git a/src/contexts/functions/timed.ts b/src/contexts/functions/timed.ts
@@ -125,7 +125,7 @@ function timed<C extends ContextTimed, P extends Array<any>>(
 ): (...params: ContextAndParameters<C, P>) => any {
   if (f instanceof utils.AsyncFunction) {
     return async (...params) => {
-      const ctx = params[0] ?? {};
+      const ctx = params[0] != null ? { ...params[0] } : {};
       const args = params.slice(1) as P;
       const teardownContext = setupTimedContext(
         delay,
@@ -140,7 +140,7 @@ function timed<C extends ContextTimed, P extends Array<any>>(
     };
   } else if (f instanceof utils.GeneratorFunction) {
     return function* (...params) {
-      const ctx = params[0] ?? {};
+      const ctx = params[0] != null ? { ...params[0] } : {};
       const args = params.slice(1) as P;
       const teardownContext = setupTimedContext(
         delay,
@@ -155,7 +155,7 @@ function timed<C extends ContextTimed, P extends Array<any>>(
     };
   } else if (f instanceof utils.AsyncGeneratorFunction) {
     return async function* (...params) {
-      const ctx = params[0] ?? {};
+      const ctx = params[0] != null ? { ...params[0] } : {};
       const args = params.slice(1) as P;
       const teardownContext = setupTimedContext(
         delay,
@@ -170,7 +170,7 @@ function timed<C extends ContextTimed, P extends Array<any>>(
     };
   } else {
     return (...params) => {
-      const ctx = params[0] ?? {};
+      const ctx = params[0] != null ? { ...params[0] } : {};
       const args = params.slice(1) as P;
       const teardownContext = setupTimedContext(
         delay,

diff --git a/src/contexts/functions/timedCancellable.ts b/src/contexts/functions/timedCancellable.ts
@@ -153,7 +153,7 @@ function timedCancellable<C extends ContextTimed, P extends Array<any>, R>(
   errorTimeoutConstructor: new () => Error = contextsErrors.ErrorContextsTimedTimeOut,
 ): (...params: ContextAndParameters<C, P>) => PromiseCancellable<R> {
   return (...params) => {
-    const ctx = params[0] ?? {};
+    const ctx = params[0] != null ? { ...params[0] } : {};
     const args = params.slice(1) as P;
     return setupTimedCancellable(
       f,

diff --git a/src/network/ConnectionForward.ts b/src/network/ConnectionForward.ts
@@ -121,7 +121,11 @@ class ConnectionForward extends Connection {
       promise<void>();
     // Promise for abortion and timeout
     const { p: abortedP, resolveP: resolveAbortedP } = promise<void>();
-    ctx.signal.addEventListener('abort', () => resolveAbortedP());
+    if (ctx.signal.aborted) {
+      resolveAbortedP();
+    } else {
+      ctx.signal.addEventListener('abort', () => resolveAbortedP());
+    }
     this.resolveReadyP = resolveReadyP;
     this.utpSocket.on('message', this.handleMessage);
     const handleStartError = (e) => {

diff --git a/src/nodes/NodeConnection.ts b/src/nodes/NodeConnection.ts
@@ -5,14 +5,17 @@ import type { Certificate, PublicKey, PublicKeyPem } from '../keys/types';
 import type Proxy from '../network/Proxy';
 import type GRPCClient from '../grpc/GRPCClient';
 import type NodeConnectionManager from './NodeConnectionManager';
-import type { Timer } from '../types';
+import type { ContextTimed } from '../contexts/types';
+import type { PromiseCancellable } from '@matrixai/async-cancellable';
 import Logger from '@matrixai/logger';
 import { CreateDestroy, ready } from '@matrixai/async-init/dist/CreateDestroy';
 import * as asyncInit from '@matrixai/async-init';
 import * as nodesErrors from './errors';
+import { context, timedCancellable } from '../contexts/index';
 import * as keysUtils from '../keys/utils';
 import * as grpcErrors from '../grpc/errors';
 import * as networkUtils from '../network/utils';
+import { timerStart } from '../utils/index';
 
 /**
  * Encapsulates the unidirectional client-side connection of one node to another.
@@ -34,31 +37,59 @@ class NodeConnection<T extends GRPCClient> {
   protected proxy: Proxy;
   protected client: T;
 
-  static async createNodeConnection<T extends GRPCClient>({
-    targetNodeId,
-    targetHost,
-    targetPort,
-    targetHostname,
-    timer,
-    proxy,
-    keyManager,
-    clientFactory,
-    nodeConnectionManager,
-    destroyCallback = async () => {},
-    logger = new Logger(this.name),
-  }: {
-    targetNodeId: NodeId;
-    targetHost: Host;
-    targetPort: Port;
-    targetHostname?: Hostname;
-    timer?: Timer;
-    proxy: Proxy;
-    keyManager: KeyManager;
-    clientFactory: (...args) => Promise<T>;
-    nodeConnectionManager: NodeConnectionManager;
-    destroyCallback?: () => Promise<void>;
-    logger?: Logger;
-  }): Promise<NodeConnection<T>> {
+  static createNodeConnection<T extends GRPCClient>(
+    {
+      targetNodeId,
+      targetHost,
+      targetPort,
+      targetHostname,
+      proxy,
+      keyManager,
+      clientFactory,
+      nodeConnectionManager,
+      destroyCallback = async () => {},
+      logger = new Logger(this.name),
+    }: {
+      targetNodeId: NodeId;
+      targetHost: Host;
+      targetPort: Port;
+      targetHostname?: Hostname;
+      proxy: Proxy;
+      keyManager: KeyManager;
+      clientFactory: (...args) => Promise<T>;
+      nodeConnectionManager: NodeConnectionManager;
+      destroyCallback?: () => Promise<void>;
+      logger?: Logger;
+    },
+    ctx?: Partial<ContextTimed>,
+  ): PromiseCancellable<NodeConnection<T>>;
+  @timedCancellable(true, 20000)
+  static async createNodeConnection<T extends GRPCClient>(
+    {
+      targetNodeId,
+      targetHost,
+      targetPort,
+      targetHostname,
+      proxy,
+      keyManager,
+      clientFactory,
+      nodeConnectionManager,
+      destroyCallback = async () => {},
+      logger = new Logger(this.name),
+    }: {
+      targetNodeId: NodeId;
+      targetHost: Host;
+      targetPort: Port;
+      targetHostname?: Hostname;
+      proxy: Proxy;
+      keyManager: KeyManager;
+      clientFactory: (...args) => Promise<T>;
+      nodeConnectionManager: NodeConnectionManager;
+      destroyCallback?: () => Promise<void>;
+      logger?: Logger;
+    },
+    @context ctx: ContextTimed,
+  ): Promise<NodeConnection<T>> {
     logger.info(`Creating ${this.name}`);
     // Checking if attempting to connect to a wildcard IP
     if (networkUtils.isHostWildcard(targetHost)) {
@@ -91,45 +122,48 @@ class NodeConnection<T extends GRPCClient> {
       destroyCallback,
       logger,
     });
-    let client;
+    let client: T;
     try {
       // Start the hole punching only if we are not connecting to seed nodes
-      let holePunchPromises: Promise<void>[] = [];
       const seedNodes = nodeConnectionManager.getSeedNodes();
       const isSeedNode = !!seedNodes.find((nodeId) => {
         return nodeId.equals(targetNodeId);
       });
       if (!isSeedNode) {
-        holePunchPromises = Array.from(seedNodes, (nodeId) => {
+        // FIXME: this needs to be cancellable.
+        //  It needs to timeout as well as abort for cleanup
+        void Array.from(seedNodes, (nodeId) => {
           return nodeConnectionManager.sendHolePunchMessage(
             nodeId,
             keyManager.getNodeId(),
             targetNodeId,
             proxyAddress,
             signature,
+            ctx,
           );
         });
       }
-      [client] = await Promise.all([
-        clientFactory({
-          nodeId: targetNodeId,
-          host: targetHost,
-          port: targetPort,
-          proxyConfig: proxyConfig,
-          // Think about this
-          logger: logger.getChild(clientFactory.name),
-          destroyCallback: async () => {
-            if (
-              nodeConnection[asyncInit.status] !== 'destroying' &&
-              !nodeConnection[asyncInit.destroyed]
-            ) {
-              await nodeConnection.destroy();
-            }
-          },
-          timer: timer,
-        }),
-        ...holePunchPromises,
-      ]);
+      // TODO: this needs to be updated to take a context,
+      //  still uses old timer style.
+      client = await clientFactory({
+        nodeId: targetNodeId,
+        host: targetHost,
+        port: targetPort,
+        proxyConfig: proxyConfig,
+        // Think about this
+        logger: logger.getChild(clientFactory.name),
+        destroyCallback: async () => {
+          if (
+            nodeConnection[asyncInit.status] !== 'destroying' &&
+            !nodeConnection[asyncInit.destroyed]
+          ) {
+            await nodeConnection.destroy();
+          }
+        },
+        // FIXME: this needs to be replaced with
+        //  the GRPC timerCancellable update
+        timer: timerStart(ctx.timer.getTimeout()),
+      });
       // 5. When finished, you have a connection to other node
       // The GRPCClient is ready to be used for requests
     } catch (e) {
@@ -142,8 +176,9 @@ class NodeConnection<T extends GRPCClient> {
       }
       throw e;
     }
+    // FIXME: we need a finally block here to do cleanup.
     // TODO: This is due to chicken or egg problem
-    // see if we can move to CreateDestroyStartStop to resolve this
+    //  see if we can move to CreateDestroyStartStop to resolve this
     nodeConnection.client = client;
     logger.info(`Created ${this.name}`);
     return nodeConnection;