Fixing intermittent failures due to js-quic stream errors

package.json

@@ -75,7 +75,7 @@
     "@matrixai/id": "^3.3.6",
     "@matrixai/logger": "^3.1.2",
     "@matrixai/mdns": "^1.2.6",
-    "@matrixai/quic": "^1.2.6",
+    "@matrixai/quic": "^1.2.7",
     "@matrixai/resources": "^1.1.5",
     "@matrixai/rpc": "^0.5.1",
     "@matrixai/timer": "^1.1.3",

src/nodes/NodeConnectionManager.ts

@@ -19,7 +19,11 @@ import {
   utils as quicUtils,
 } from '@matrixai/quic';
 import { withF } from '@matrixai/resources';
-import { middleware as rpcMiddleware, RPCServer } from '@matrixai/rpc';
+import {
+  errors as rpcErrors,
+  middleware as rpcMiddleware,
+  RPCServer,
+} from '@matrixai/rpc';
 import Logger from '@matrixai/logger';
 import { Timer } from '@matrixai/timer';
 import { IdInternal } from '@matrixai/id';
@@ -850,14 +854,23 @@ class NodeConnectionManager {
           this.keyRing.keyPair,
           data,
         );
-        const addressMessage =
-          await client.methods.nodesConnectionSignalInitial(
+        const addressMessage = await client.methods
+          .nodesConnectionSignalInitial(
             {
               targetNodeIdEncoded: nodesUtils.encodeNodeId(nodeIdTarget),
               signature: signature.toString('base64url'),
             },
             ctx,
-          );
+          )
+          .catch((e) => {
+            if (e instanceof rpcErrors.ErrorRPCHandlerFailed) {
+              throw new nodesErrors.ErrorNodeConnectionManagerSignalFailed(
+                'Failed initial signal step triggering `nodesConnectionSignalInitial`',
+                { cause: e },
+              );
+            }
+            throw e;
+          });
         return {
           host: addressMessage.host as Host,
           port: addressMessage.port as Port,
@@ -940,7 +953,7 @@ class NodeConnectionManager {
   /**
    * Gets the existing active connection for the target node
    */
-  public getConnection(nodeId): ConnectionAndTimer | undefined {
+  public getConnection(nodeId: NodeId): ConnectionAndTimer | undefined {
     const nodeIdString = nodeId.toString() as NodeIdString;
     const connectionsEntry = this.connections.get(nodeIdString);
     if (connectionsEntry == null) return;
@@ -1292,7 +1305,17 @@ class NodeConnectionManager {
       .then(
         () => {},
         (e) => {
-          if (!nodesUtils.isConnectionError(e)) throw e;
+          // If it's a connection error or missing handler then it's a signalling failure
+          if (
+            nodesUtils.isConnectionError(e) ||
+            e instanceof rpcErrors.ErrorRPCHandlerFailed
+          ) {
+            new nodesErrors.ErrorNodeConnectionManagerSignalFailed(
+              'Failed final signal step triggering `nodesConnectionSignalFinal`',
+              { cause: e },
+            );
+          }
+          throw e;
         },
       )
       .finally(() => {

src/nodes/errors.ts

@@ -188,6 +188,13 @@ class ErrorNodeConnectionManagerRequestRateExceeded<
   exitCode = sysexits.TEMPFAIL;
 }
 
+class ErrorNodeConnectionManagerSignalFailed<
+  T,
+> extends ErrorNodeConnectionManager<T> {
+  static description = 'Failed to signal hole punching';
+  exitCode = sysexits.TEMPFAIL;
+}
+
 class ErrorNodePingFailed<T> extends ErrorNodes<T> {
   static description =
     'Failed to ping the node when attempting to authenticate';
@@ -240,6 +247,7 @@ export {
   ErrorNodeConnectionManagerMultiConnectionFailed,
   ErrorNodeConnectionManagerConnectionNotFound,
   ErrorNodeConnectionManagerRequestRateExceeded,
+  ErrorNodeConnectionManagerSignalFailed,
   ErrorNodePingFailed,
   ErrorNodePermissionDenied,
   ErrorNodeLookupNotFound,

src/vaults/VaultInternal.ts

@@ -816,6 +816,7 @@ class VaultInternal {
       vaultNameOrId: vaultNameOrId_,
       action: vaultAction,
     });
+
     const result = vaultsGitInfoGetStream.meta?.result;
     if (result == null || !utils.isObject(result)) {
       utils.never('`result` must be a defined object');
@@ -832,6 +833,11 @@ class VaultInternal {
     const vaultName = result.vaultName;
     const remoteVaultId = ids.parseVaultId(result.vaultIdEncoded);
 
+    const vaultsGitPackGetStream = await client.methods.vaultsGitPackGet({
+      nameOrId: result.vaultIdEncoded as string,
+      vaultAction,
+    });
+
     return [
       async function ({
         url,
@@ -855,14 +861,9 @@ class VaultInternal {
             statusMessage: 'OK',
           };
         } else if (method === 'POST') {
-          const vaultsGitPackGetStream = await client.methods.vaultsGitPackGet({
-            nameOrId: result.vaultIdEncoded as string,
-            vaultAction,
-          });
           const writer = vaultsGitPackGetStream.writable.getWriter();
           await writer.write(body[0]);
           await writer.close();
-
           return {
             url: url,
             method: method,

tests/vaults/VaultManager.test.ts

@@ -928,7 +928,6 @@ describe('VaultManager', () => {
         vaultsErrors.ErrorVaultsPermissionDenied,
       );
     });
-    // FIXME: Test has been disabled due to non-deterministic failures in CI/CD
     test(
       'can pull a cloned vault',
       async () => {