tests: reduced memory usage of the client domain tests

Author: tegefaulkes

package.json

@@ -51,7 +51,7 @@
     "postversion": "npm install --package-lock-only --ignore-scripts --silent",
     "ts-node": "ts-node",
     "ts-node-inspect": "node --require ts-node/register --inspect",
-    "test": "jest",
+    "test": "node --expose-gc ./node_modules/.bin/jest",
     "lint": "eslint '{src,tests,scripts}/**/*.{js,ts,json}' 'benches/**/*.{js,ts}'",
     "lintfix": "eslint '{src,tests,scripts}/**/*.{js,ts,json}' 'benches/**/*.{js,ts}' --fix",
     "lint-shell": "find ./src ./tests ./scripts -type f -regextype posix-extended -regex '.*\\.(sh)' -exec shellcheck {} +",

shell.nix

@@ -9,8 +9,6 @@ in
       nodejs
       shellcheck
       gitAndTools.gh
-      awscli2
-      skopeo
       jq
     ];
     PKG_CACHE_PATH = utils.pkgCachePath;

src/nodes/NodeConnectionManager.ts

@@ -697,7 +697,7 @@ class NodeConnectionManager {
       ctx.signal.removeEventListener('abort', onAbort);
     };
     ctx.signal.addEventListener('abort', onAbort);
-    void ctx.timer.catch(() => {}).finally(() => onAbort());
+    const timer = ctx.timer.catch(() => {}).finally(() => onAbort());
     let delay = this.connectionHolePunchIntervalTime;
     // Setting up established event checking
     try {
@@ -710,6 +710,7 @@ class NodeConnectionManager {
       }
     } finally {
       onAbort();
+      await timer;
     }
   }
 

src/rpc/RPCClient.ts

@@ -1,5 +1,5 @@
 import type { WritableStream, ReadableStream } from 'stream/web';
-import type { ContextTimed } from '@matrixai/contexts';
+import type { ContextTimed, ContextTimedInput } from '@matrixai/contexts';
 import type {
   HandlerType,
   JSONRPCRequestMessage,
@@ -156,7 +156,7 @@ class RPCClient<M extends ClientManifest> {
   public async unaryCaller<I extends JSONValue, O extends JSONValue>(
     method: string,
     parameters: I,
-    ctx: Partial<ContextTimed> = {},
+    ctx: Partial<ContextTimedInput> = {},
   ): Promise<O> {
     const callerInterface = await this.duplexStreamCaller<I, O>(method, ctx);
     const reader = callerInterface.readable.getReader();
@@ -191,7 +191,7 @@ class RPCClient<M extends ClientManifest> {
   public async serverStreamCaller<I extends JSONValue, O extends JSONValue>(
     method: string,
     parameters: I,
-    ctx: Partial<ContextTimed> = {},
+    ctx: Partial<ContextTimedInput> = {},
   ): Promise<ReadableStream<O>> {
     const callerInterface = await this.duplexStreamCaller<I, O>(method, ctx);
     const writer = callerInterface.writable.getWriter();
@@ -219,7 +219,7 @@ class RPCClient<M extends ClientManifest> {
   @ready(new rpcErrors.ErrorRPCDestroyed())
   public async clientStreamCaller<I extends JSONValue, O extends JSONValue>(
     method: string,
-    ctx: Partial<ContextTimed> = {},
+    ctx: Partial<ContextTimedInput> = {},
   ): Promise<{
     output: Promise<O>;
     writable: WritableStream<I>;
@@ -252,7 +252,7 @@ class RPCClient<M extends ClientManifest> {
   @ready(new rpcErrors.ErrorRPCDestroyed())
   public async duplexStreamCaller<I extends JSONValue, O extends JSONValue>(
     method: string,
-    ctx: Partial<ContextTimed> = {},
+    ctx: Partial<ContextTimedInput> = {},
   ): Promise<RPCStream<O, I>> {
     const abortController = new AbortController();
     const signal = abortController.signal;
@@ -270,11 +270,14 @@ class RPCClient<M extends ClientManifest> {
       if (ctx.signal.aborted) abortHandler();
       ctx.signal.addEventListener('abort', abortHandler);
     }
-    const timer =
-      ctx.timer ??
-      new Timer({
-        delay: this.streamKeepAliveTimeoutTime,
+    let timer: Timer;
+    if (!(ctx.timer instanceof Timer)) {
+      timer = new Timer({
+        delay: ctx.timer ?? this.streamKeepAliveTimeoutTime,
       });
+    } else {
+      timer = ctx.timer;
+    }
     const cleanUp = () => {
       // Clean up the timer and signal
       if (ctx.timer == null) timer.cancel(timerCleanupReasonSymbol);

src/rpc/types.ts

@@ -1,5 +1,5 @@
 import type { ReadableStream, ReadableWritablePair } from 'stream/web';
-import type { ContextTimed } from '@matrixai/contexts';
+import type { ContextTimed, ContextTimedInput } from '@matrixai/contexts';
 import type { Handler } from './handlers';
 import type {
   Caller,
@@ -239,28 +239,31 @@ type MiddlewareFactory<FR, FW, RR, RW> = (
 type UnaryCallerImplementation<
   I extends JSONValue = JSONValue,
   O extends JSONValue = JSONValue,
-> = (parameters: I, ctx?: Partial<ContextTimed>) => Promise<O>;
+> = (parameters: I, ctx?: Partial<ContextTimedInput>) => Promise<O>;
 
 type ServerCallerImplementation<
   I extends JSONValue = JSONValue,
   O extends JSONValue = JSONValue,
-> = (parameters: I, ctx?: Partial<ContextTimed>) => Promise<ReadableStream<O>>;
+> = (
+  parameters: I,
+  ctx?: Partial<ContextTimedInput>,
+) => Promise<ReadableStream<O>>;
 
 type ClientCallerImplementation<
   I extends JSONValue = JSONValue,
   O extends JSONValue = JSONValue,
 > = (
-  ctx?: Partial<ContextTimed>,
+  ctx?: Partial<ContextTimedInput>,
 ) => Promise<{ output: Promise<O>; writable: WritableStream<I> }>;
 
 type DuplexCallerImplementation<
   I extends JSONValue = JSONValue,
   O extends JSONValue = JSONValue,
-> = (ctx?: Partial<ContextTimed>) => Promise<RPCStream<O, I>>;
+> = (ctx?: Partial<ContextTimedInput>) => Promise<RPCStream<O, I>>;
 
 type RawCallerImplementation = (
   headerParams: JSONValue,
-  ctx?: Partial<ContextTimed>,
+  ctx?: Partial<ContextTimedInput>,
 ) => Promise<RPCStream<Uint8Array, Uint8Array>>;
 
 type ConvertDuplexCaller<T> = T extends DuplexCaller<infer I, infer O>

tests/agent/handlers/nodesHolePunchMessage.test.ts

@@ -221,7 +221,6 @@ describe('nodesHolePunchMessage', () => {
     await nodeGraph.stop();
     await nodeManager.stop();
     await nodeConnectionManager.stop();
-    await quicSocket.stop();
     await taskManager.stop();
     await sigchain.stop();
     await acl.stop();