chore: resolved lint and failing NodeManager tests

Author: aryanjassal

src/discovery/Discovery.ts

@@ -153,13 +153,11 @@ class Discovery {
     parent: GestaltIdEncoded | null,
   ) => {
     try {
-      this.logger.error('out here 1')
       await this.processVertex(
         vertex,
         lastProcessedCutoffTime ?? undefined,
         ctx,
       );
-      this.logger.error('out here 2')
       this.dispatchEvent(
         new discoveryEvents.EventDiscoveryVertexProcessed({
           detail: {
@@ -171,13 +169,11 @@ class Discovery {
     } catch (e) {
       // We need to reschedule if the task was cancelled due to discovery domain stopping
       if (e === discoveryStoppingTaskReason) {
-        this.logger.error('out here 3')
         // We need to recreate the task for the vertex
         const vertexId = gestaltsUtils.decodeGestaltId(vertex);
         if (vertexId == null) {
           never(`failed to decode vertex GestaltId "${vertex}"`);
         }
-        this.logger.error('out here 4')
         await this.scheduleDiscoveryForVertex(
           vertexId,
           undefined,
@@ -187,13 +183,14 @@ class Discovery {
           undefined,
           ctx,
         );
-        this.logger.error('out here 5')
         return;
       }
       // Aborting a duplicate task is not an error
       if (e === abortSingletonTaskReason) return;
       // Destroying tasks is not an error
       if (e === discoveryDestroyedTaskReason) return;
+      // Abortion due to cancellation is not an error
+      if (e === ctx.signal.reason) return;
       this.dispatchEvent(
         new discoveryEvents.EventDiscoveryVertexFailed({
           detail: {
@@ -213,7 +210,9 @@ class Discovery {
   /**
    * This handler is run periodically to check if nodes are ready to be rediscovered
    */
-  protected checkRediscoveryHandler: TaskHandler = async (ctx: ContextTimed) => {
+  protected checkRediscoveryHandler: TaskHandler = async (
+    ctx: ContextTimed,
+  ) => {
     await this.checkRediscovery(
       Date.now() - this.rediscoverVertexThresholdTime,
       undefined,
@@ -368,7 +367,7 @@ class Discovery {
     providerId: ProviderId,
     identityId: IdentityId,
     lastProcessedCutoffTime?: number,
-    ctx?: Partial<ContextTimedInput>
+    ctx?: Partial<ContextTimedInput>,
   ): Promise<void>;
   @ready(new discoveryErrors.ErrorDiscoveryNotRunning())
   @timedCancellable(true)
@@ -439,17 +438,11 @@ class Discovery {
     const [type, id] = vertexId;
     switch (type) {
       case 'node':
-        this.logger.error('processnode before')
-        // return await this.processNode(id, lastProcessedCutoffTime, ctx);
-        const val1 = await this.processNode(id, lastProcessedCutoffTime, ctx);
-        this.logger.error('processnode after')
-        return val1
+        // Return await this.processNode(id, lastProcessedCutoffTime, ctx);
+        return await this.processNode(id, lastProcessedCutoffTime, ctx);
       case 'identity':
-        this.logger.error('processidentity before')
-        // return await this.processIdentity(id, lastProcessedCutoffTime, ctx);
-        const val2 = await this.processIdentity(id, lastProcessedCutoffTime, ctx);
-        this.logger.error('processidentity after')
-        return val2
+        // Return await this.processIdentity(id, lastProcessedCutoffTime, ctx);
+        return await this.processIdentity(id, lastProcessedCutoffTime, ctx);
       default:
         never(`type must be either "node" or "identity" got "${type}"`);
     }
@@ -466,34 +459,26 @@ class Discovery {
     if (nodeId.equals(this.keyRing.getNodeId())) {
       // Skip our own nodeId, we actively add this information when it changes,
       // so there is no need to scan it.
-      this.logger.error('before processed tiem')
       await this.gestaltGraph.setVertexProcessedTime(
         gestaltNodeId,
         processedTime,
       );
-      this.logger.error('after processed tiem')
       return;
     }
-    this.logger.error('before claim')
     const newestClaimId = await this.gestaltGraph.getClaimIdNewest(nodeId);
-    this.logger.error('after claim')
     // The sigChain data of the vertex (containing all cryptolinks)
     let vertexChainData: Record<ClaimIdEncoded, SignedClaim> = {};
     try {
-      this.logger.error('before chain')
       vertexChainData = await this.nodeManager.requestChainData(
         nodeId,
         newestClaimId,
         ctx,
       );
-      this.logger.error('after chain')
     } catch (e) {
-      this.logger.error('before chain time error')
       await this.gestaltGraph.setVertexProcessedTime(
         gestaltNodeId,
         processedTime,
       );
-      this.logger.error('after chain time error')
       // Not strictly an error in this case, we can fail to connect
       this.logger.info(
         `Failed to discover ${nodesUtils.encodeNodeId(
@@ -507,37 +492,31 @@ class Discovery {
       ctx.signal.throwIfAborted();
       switch (signedClaim.payload.typ) {
         case 'ClaimLinkNode':
-          this.logger.error('claimlinknode before')
           await this.processClaimLinkNode(
             signedClaim as SignedClaim<ClaimLinkNode>,
             nodeId,
             lastProcessedCutoffTime,
             ctx,
           );
-          this.logger.error('claimlinknode after')
           break;
         case 'ClaimLinkIdentity':
-          this.logger.error('claimlinkidentity before')
           await this.processClaimLinkIdentity(
             signedClaim as SignedClaim<ClaimLinkIdentity>,
             nodeId,
             lastProcessedCutoffTime,
             ctx,
           );
-          this.logger.error('claimlinkidentity after')
           break;
         default:
           never(
             `signedClaim.payload.typ must be "ClaimLinkNode" or "ClaimLinkIdentity" got "${signedClaim.payload.typ}"`,
           );
       }
     }
-    this.logger.error('setvertex time before')
     await this.gestaltGraph.setVertexProcessedTime(
       gestaltNodeId,
       processedTime,
     );
-    this.logger.error('setvertex time after')
   }
 
   protected async processClaimLinkNode(
@@ -549,7 +528,6 @@ class Discovery {
     // Get the chain data of the linked node
     // Could be node1 or node2 in the claim so get the one that's
     // not equal to nodeId from above
-    this.logger.error('processClaimLinkNode 1')
     const node1Id = nodesUtils.decodeNodeId(signedClaim.payload.iss);
     if (node1Id == null) {
       never(`failed to decode issuer NodeId "${signedClaim.payload.iss}"`);
@@ -558,7 +536,6 @@ class Discovery {
     if (node2Id == null) {
       never(`failed to decode subject NodeId "${signedClaim.payload.sub}"`);
     }
-    this.logger.error('processClaimLinkNode 2')
     // Verify the claim
     const node1PublicKey = keysUtils.publicKeyFromNodeId(node1Id);
     const node2PublicKey = keysUtils.publicKeyFromNodeId(node2Id);
@@ -572,12 +549,10 @@ class Discovery {
       );
       return;
     }
-    this.logger.error('processClaimLinkNode 3')
     const linkedNodeId = node1Id.equals(nodeId) ? node2Id : node1Id;
     const linkedVertexNodeInfo: GestaltNodeInfo = {
       nodeId: linkedNodeId,
     };
-    this.logger.error('processClaimLinkNode 4')
     await this.gestaltGraph.linkNodeAndNode(
       {
         nodeId,
@@ -588,7 +563,6 @@ class Discovery {
         meta: {},
       },
     );
-    this.logger.error('processClaimLinkNode 5')
     const claimId = decodeClaimId(signedClaim.payload.jti);
     if (claimId == null) {
       never(`failed to decode claimId "${signedClaim.payload.jti}"`);
@@ -602,7 +576,6 @@ class Discovery {
         lastProcessedCutoffTime,
       ))
     ) {
-      this.logger.error('processClaimLinkNode 7')
       await this.scheduleDiscoveryForVertex(
         linkedGestaltId,
         undefined,
@@ -884,7 +857,6 @@ class Discovery {
         gestaltIdEncoded,
       ].join(''),
     );
-    this.logger.error('here1')
     // Check if task exists
     let taskExisting: Task | null = null;
     for await (const task of this.taskManager.getTasks(
@@ -895,14 +867,12 @@ class Discovery {
       ctx,
     )) {
       ctx.signal.throwIfAborted();
-      this.logger.error('here2')
       // Ignore active tasks
       if (ignoreActive && task.status === 'active') continue;
       if (taskExisting == null) {
         taskExisting = task;
         continue;
       }
-      this.logger.error('here3')
       // Any extra tasks should be cancelled, this shouldn't normally happen
       task.cancel(abortSingletonTaskReason);
       this.dispatchEvent(
@@ -914,13 +884,11 @@ class Discovery {
         }),
       );
     }
-    this.logger.error('here4')
     // Only create if it doesn't exist
     if (taskExisting != null) return;
     this.logger.info(
       `Scheduling new discovery for vertex with gestaltId ${gestaltIdEncoded}`,
     );
-    this.logger.error('here5')
     await this.taskManager.scheduleTask(
       {
         handlerId: this.discoverVertexHandlerId,
@@ -936,7 +904,6 @@ class Discovery {
       },
       tran,
     );
-    this.logger.error('here6')
     this.dispatchEvent(
       new discoveryEvents.EventDiscoveryVertexQueued({
         detail: {
@@ -1087,12 +1054,10 @@ class Discovery {
           }
           // Refresh timer in preparation for request
           ctx.timer.refresh();
-          this.logger.error('verifyIdentityClaim before getClaim')
           const identitySignedClaim = await provider.getClaim(
             authIdentityId,
             claimId,
           );
-          this.logger.error('verifyIdentityClaim after getClaim')
           if (identitySignedClaim == null) {
             continue;
           }
@@ -1119,10 +1084,11 @@ class Discovery {
     tran?: DBTransaction,
     ctx?: Partial<ContextTimedInput>,
   ): Promise<void>;
+  @timedCancellable(true)
   public async checkRediscovery(
     lastProcessedCutoffTime: number,
     tran: DBTransaction | undefined,
-    ctx: ContextTimed,
+    @context ctx: ContextTimed,
   ): Promise<void> {
     if (tran == null) {
       return this.db.withTransactionF((tran) =>

src/nodes/NodeConnectionManager.ts

@@ -46,8 +46,8 @@ import { AbstractEvent, EventAll } from '@matrixai/events';
 import {
   context,
   timed,
-  timedCancellable
-} from "@matrixai/contexts/dist/decorators";
+  timedCancellable,
+} from '@matrixai/contexts/dist/decorators';
 import { Semaphore } from '@matrixai/async-locks';
 import { PromiseCancellable } from '@matrixai/async-cancellable';
 import NodeConnection from './NodeConnection';
@@ -1479,7 +1479,11 @@ class NodeConnectionManager {
     port: Port;
   }>;
   @ready(new nodesErrors.ErrorNodeManagerNotRunning())
-  @timedCancellable(true)
+  @timedCancellable(
+    true,
+    (nodeConnectionManager: NodeConnectionManager) =>
+      nodeConnectionManager.connectionConnectTimeoutTime,
+  )
   public async handleNodesConnectionSignalInitial(
     sourceNodeId: NodeId,
     targetNodeId: NodeId,
@@ -1518,16 +1522,20 @@ class NodeConnectionManager {
       this.keyRing.keyPair,
       data,
     );
-    const connectionSignalP = this.withConnF(targetNodeId, ctx, async (conn) => {
-      const client = conn.getClient();
-      await client.methods.nodesConnectionSignalFinal({
-        sourceNodeIdEncoded: nodesUtils.encodeNodeId(sourceNodeId),
-        targetNodeIdEncoded: nodesUtils.encodeNodeId(targetNodeId),
-        address: address,
-        requestSignature: requestSignature,
-        relaySignature: relaySignature.toString('base64url'),
-      });
-    })
+    const connectionSignalP = this.withConnF(
+      targetNodeId,
+      ctx,
+      async (conn) => {
+        const client = conn.getClient();
+        await client.methods.nodesConnectionSignalFinal({
+          sourceNodeIdEncoded: nodesUtils.encodeNodeId(sourceNodeId),
+          targetNodeIdEncoded: nodesUtils.encodeNodeId(targetNodeId),
+          address: address,
+          requestSignature: requestSignature,
+          relaySignature: relaySignature.toString('base64url'),
+        });
+      },
+    )
       // Ignore results and failures, then are expected to happen and are allowed
       .then(
         () => {},
@@ -1790,25 +1798,26 @@ class NodeConnectionManager {
     nodeId: NodeId,
     ctx?: Partial<ContextTimedInput>,
   ): Promise<void>;
-  @timedCancellable(true)
+  @timedCancellable(
+    true,
+    (nodeConnectionManager: NodeConnectionManager) =>
+      nodeConnectionManager.connectionConnectTimeoutTime,
+  )
   public async isAuthenticatedP(
     nodeId: NodeId,
     @context ctx: ContextTimed,
   ): Promise<void> {
+    ctx.signal.throwIfAborted();
     const targetNodeIdString = nodeId.toString() as NodeIdString;
     const connectionsEntry = this.connections.get(targetNodeIdString);
     if (connectionsEntry == null) {
       throw new nodesErrors.ErrorNodeConnectionManagerConnectionNotFound();
     }
-    const { p: abortP, rejectP: triggerAbort } = utils.promise<void>();
+    const { p: abortP, rejectP: rejectAbortP } = utils.promise<never>();
     const abortHandler = () => {
-      triggerAbort(ctx.signal.reason);
+      rejectAbortP(ctx.signal.reason);
     };
-    if (ctx.signal.aborted) {
-      triggerAbort(ctx.signal.reason);
-    } else {
-      ctx.signal.addEventListener('abort', abortHandler, { once: true });
-    }
+    ctx.signal.addEventListener('abort', abortHandler, { once: true });
     try {
       return await Promise.race([connectionsEntry.authenticatedP, abortP]);
     } catch (e) {