feat: reimplementing hole punch logic

* Related #527

[ci skip]

Author: tegefaulkes

src/nodes/NodeConnectionManager.ts

@@ -21,14 +21,14 @@ import type { RPCStream } from '../rpc/types';
 import type { TLSConfig } from '../network/types';
 import type { QuicConfig } from './types';
 import type { ServerCrypto, events as QuicEvents } from '@matrixai/quic';
+import type { PromiseCancellable } from '@matrixai/async-cancellable';
 import { withF } from '@matrixai/resources';
 import Logger from '@matrixai/logger';
 import { ready, StartStop } from '@matrixai/async-init/dist/StartStop';
 import { IdInternal } from '@matrixai/id';
 import { Lock, LockBox } from '@matrixai/async-locks';
 import { Timer } from '@matrixai/timer';
 import { timedCancellable, context } from '@matrixai/contexts/dist/decorators';
-import { PromiseCancellable } from '@matrixai/async-cancellable';
 import { QUICServer } from '@matrixai/quic';
 import NodeConnection from './NodeConnection';
 import * as nodesUtils from './utils';
@@ -598,7 +598,7 @@ class NodeConnectionManager {
     this.logger.debug(
       `establishing single connection for address ${address.host}:${address.port}`,
     );
-    const iceProm = this.initiateIce();
+    const iceProm = this.initiateHolePunch(nodeIds, ctx);
     const connection =
       await NodeConnection.createNodeConnection<AgentClientManifest>(
         {
@@ -624,7 +624,7 @@ class NodeConnectionManager {
           throw e;
         })
         .finally(async () => {
-          iceProm.cancel();
+          iceProm.cancel('Connection was established');
           await iceProm;
         });
     // 2. if established then add to result map
@@ -1425,11 +1425,50 @@ class NodeConnectionManager {
     this.nodesBackoffMap.delete(nodeId.toString());
   }
 
-  protected initiateIce(): PromiseCancellable<void> {
-    // TODO: this is a placeholder for ICE operation
-    return new PromiseCancellable<void>((resolve) => {
-      resolve();
-    });
+  /**
+   * This attempts the NAT hole punch procedure. It will return a
+   * `PromiseCancellable` that will resolve once the procedure times out, is
+   * cancelled or the other end responds.
+   *
+   * This is pretty simple, it will contact all known seed nodes and get them to
+   * relay a punch signal message.
+   *
+   * Note: Avoid using a large set of target nodes, It could trigger a large
+   * amount of pings to a single target.
+   */
+  protected initiateHolePunch(
+    targetNodeIds: Array<NodeId>,
+    ctx?: Partial<ContextTimedInput>,
+  ): PromiseCancellable<void>;
+  @timedCancellable(true)
+  protected async initiateHolePunch(
+    targetNodeIds: Array<NodeId>,
+    @context ctx: ContextTimed,
+  ): Promise<void> {
+    const seedNodes = this.getSeedNodes();
+    const allProms: Array<Promise<Array<void>>> = [];
+    for (const targetNodeId of targetNodeIds) {
+      if (!this.isSeedNode(targetNodeId)) {
+        const holePunchProms = seedNodes.map((seedNodeId) => {
+          return (
+            this.sendSignalingMessage(
+              seedNodeId,
+              this.keyRing.getNodeId(),
+              targetNodeId,
+              undefined,
+              ctx,
+            )
+              // Ignore results
+              .then(
+                () => {},
+                () => {},
+              )
+          );
+        });
+        allProms.push(Promise.all(holePunchProms));
+      }
+    }
+    await Promise.all(allProms).catch();
   }
 }
 

src/rpc/RPCClient.ts

@@ -257,15 +257,12 @@ class RPCClient<M extends ClientManifest> {
     const abortController = new AbortController();
     const signal = abortController.signal;
     // A promise that will reject if there is an abort signal or timeout
-    const abortRaceProm = promise<never>();
     // Prevent unhandled rejection when we're done with the promise
-    abortRaceProm.p.catch(() => {});
     let abortHandler: () => void;
     if (ctx.signal != null) {
       // Propagate signal events
       abortHandler = () => {
         abortController.abort(ctx.signal?.reason);
-        abortRaceProm.rejectP(ctx.signal?.reason);
       };
       if (ctx.signal.aborted) abortHandler();
       ctx.signal.addEventListener('abort', abortHandler);
@@ -288,21 +285,25 @@ class RPCClient<M extends ClientManifest> {
     void timer.then(
       () => {
         abortController.abort(timeoutError);
-        abortRaceProm.rejectP(timeoutError);
       },
       () => {}, // Ignore cancellation error
     );
     // Hooking up agnostic stream side
     let rpcStream: RPCStream<Uint8Array, Uint8Array>;
     try {
-      rpcStream = await Promise.race([
-        this.streamFactory({ signal, timer }),
-        abortRaceProm.p,
-      ]);
+      rpcStream = await this.streamFactory({ signal, timer });
     } catch (e) {
       cleanUp();
       throw e;
     }
+    const cancelStream = () => {
+      rpcStream.cancel(signal.reason);
+    };
+    if (signal.aborted) {
+      cancelStream();
+    } else {
+      signal.addEventListener('abort', cancelStream);
+    }
     // Setting up event for stream timeout
     void timer.then(
       () => {
@@ -345,6 +346,7 @@ class RPCClient<M extends ClientManifest> {
         .catch(() => {}),
     ]).finally(() => {
       cleanUp();
+      signal.removeEventListener('abort', cancelStream);
     });
 
     // Returning interface

src/rpc/RPCServer.ts

@@ -1,3 +1,4 @@
+import type { ReadableStreamDefaultReadResult } from 'stream/web';
 import type {
   ClientHandlerImplementation,
   DuplexHandlerImplementation,
@@ -442,7 +443,6 @@ class RPCServer extends EventTarget {
         });
     };
     abortController.signal.addEventListener('abort', handleAbort);
-
     const prom = (async () => {
       const headTransformStream = rpcUtilsMiddleware.binaryToJsonMessageStream(
         rpcUtils.parseJSONRPCRequest,
@@ -469,30 +469,80 @@ class RPCServer extends EventTarget {
         await inputStream.cancel(reason);
         await rpcStream.writable.abort(reason);
         await inputStreamEndProm;
+        timer.cancel(cleanupReason);
+        graceTimer?.cancel(cleanupReason);
+        await timer.catch(() => {});
+        await graceTimer?.catch(() => {});
       };
       // Read a single empty value to consume the first message
       const reader = headTransformStream.readable.getReader();
       // Allows timing out when waiting for the first message
-      const headerMessage = await Promise.race([
-        reader.read(),
-        timer.then(
-          () => undefined,
-          () => {},
-        ),
-      ]);
+      let headerMessage:
+        | ReadableStreamDefaultReadResult<JSONRPCRequest>
+        | undefined
+        | void;
+      try {
+        headerMessage = await Promise.race([
+          reader.read(),
+          timer.then(
+            () => undefined,
+            () => {},
+          ),
+        ]);
+      } catch (e) {
+        const newErr = new rpcErrors.ErrorRPCHandlerFailed(
+          'Stream failed waiting for header',
+          { cause: e },
+        );
+        await inputStreamEndProm;
+        timer.cancel(cleanupReason);
+        graceTimer?.cancel(cleanupReason);
+        await timer.catch(() => {});
+        await graceTimer?.catch(() => {});
+        this.dispatchEvent(
+          new rpcEvents.RPCErrorEvent({
+            detail: new rpcErrors.ErrorRPCOutputStreamError(
+              'Stream failed waiting for header',
+              {
+                cause: newErr,
+              },
+            ),
+          }),
+        );
+        return;
+      }
       // Downgrade back to the raw stream
       await reader.cancel();
       // There are 2 conditions where we just end here
       //  1. The timeout timer resolves before the first message
       //  2. the stream ends before the first message
       if (headerMessage == null) {
-        await cleanUp(
-          new rpcErrors.ErrorRPCHandlerFailed('Timed out waiting for header'),
+        const newErr = new rpcErrors.ErrorRPCHandlerFailed(
+          'Timed out waiting for header',
+        );
+        await cleanUp(newErr);
+        this.dispatchEvent(
+          new rpcEvents.RPCErrorEvent({
+            detail: new rpcErrors.ErrorRPCOutputStreamError(
+              'Timed out waiting for header',
+              {
+                cause: newErr,
+              },
+            ),
+          }),
         );
         return;
       }
       if (headerMessage.done) {
-        await cleanUp(new rpcErrors.ErrorRPCHandlerFailed('Missing header'));
+        const newErr = new rpcErrors.ErrorRPCHandlerFailed('Missing header');
+        await cleanUp(newErr);
+        this.dispatchEvent(
+          new rpcEvents.RPCErrorEvent({
+            detail: new rpcErrors.ErrorRPCOutputStreamError('Missing header', {
+              cause: newErr,
+            }),
+          }),
+        );
         return;
       }
       const method = headerMessage.value.method;
@@ -514,6 +564,7 @@ class RPCServer extends EventTarget {
         // Otherwise refresh
         timer.refresh();
       }
+      this.logger.info(`Handling stream with method (${method})`);
       const outputStream = handler(
         [headerMessage.value, inputStream],
         rpcStream.cancel,
@@ -524,6 +575,7 @@ class RPCServer extends EventTarget {
         .pipeTo(rpcStream.writable)
         .catch(() => {}); // Ignore any errors, we only care that it finished
       await Promise.allSettled([inputStreamEndProm, outputStreamEndProm]);
+      this.logger.info(`Handled stream with method (${method})`);
       // Cleaning up abort and timer
       timer.cancel(cleanupReason);
       abortController.signal.removeEventListener('abort', handleAbort);

tests/nodes/NodeConnectionManager.seednodes.test.ts

@@ -21,7 +21,6 @@ import Sigchain from '@/sigchain/Sigchain';
 import TaskManager from '@/tasks/TaskManager';
 import NodeManager from '@/nodes/NodeManager';
 import PolykeyAgent from '@/PolykeyAgent';
-import { sleep } from '@/utils';
 import * as testNodesUtils from './utils';
 import * as tlsTestUtils from '../utils/tls';
 
@@ -314,7 +313,6 @@ describe(`${NodeConnectionManager.name} seednodes test`, () => {
     await remotePolykeyAgent1.nodeGraph.setNode(remoteNodeId2, remoteAddress2);
 
     await nodeManager.syncNodeGraph(true, 100);
-    await sleep(1000);
     expect(mockedRefreshBucket).toHaveBeenCalled();
 
     await nodeConnectionManager.stop();

tests/nodes/NodeManager.test.ts

@@ -382,7 +382,8 @@ describe(`${NodeManager.name} test`, () => {
     await nodeConnectionManager.withConnF(serverNodeId, async () => {
       // Do nothing
     });
-
+    // Wait for background logic to settle
+    await sleep(100);
     const nodeData2 = await server.nodeGraph.getNode(expectedNodeId);
     expect(nodeData2).toBeDefined();
     expect(nodeData2?.address.host).toEqual(expectedHost);