fix: fixed promise leak and event emitter warnings

Author: tegefaulkes

src/nodes/NodeConnectionQueue.ts

@@ -188,29 +188,35 @@ export class NodeConnectionQueue {
     queue: Array<unknown>,
     ctx: ContextCancellable,
   ): Promise<void> {
-    const abortP = utils.signalPromise(ctx.signal).catch(() => {});
-    while (
-      !this.connectionMade &&
-      queue.length === 0 &&
-      (this.nodesRunningSignal.size > 0 ||
-        this.nodesRunningDirect.size > 0 ||
-        this.queueSignal.length > 0 ||
-        this.queueDirect.length > 0)
-    ) {
-      if (ctx.signal.aborted) return;
-      if (this.nodesRunningSignal.size + this.nodesRunningDirect.size === 0) {
-        // Yield to the event loop to allow queued attempts to start
-        await utils.sleep(0);
-        continue;
-      }
-      const runningPs: Array<Promise<void>> = [];
-      for (const P of this.nodesRunningSignal) {
-        runningPs.push(P);
-      }
-      for (const P of this.nodesRunningDirect) {
-        runningPs.push(P);
+    const abortP = utils.signalPromise(ctx.signal);
+    try {
+      while (
+        !this.connectionMade &&
+        queue.length === 0 &&
+        (this.nodesRunningSignal.size > 0 ||
+          this.nodesRunningDirect.size > 0 ||
+          this.queueSignal.length > 0 ||
+          this.queueDirect.length > 0)
+      ) {
+        if (ctx.signal.aborted) return;
+        if (this.nodesRunningSignal.size + this.nodesRunningDirect.size === 0) {
+          // Yield to the event loop to allow queued attempts to start
+          await utils.sleep(0);
+          continue;
+        }
+        const runningPs: Array<Promise<void>> = [];
+        for (const P of this.nodesRunningSignal) {
+          runningPs.push(P);
+        }
+        for (const P of this.nodesRunningDirect) {
+          runningPs.push(P);
+        }
+        await Promise.any([...runningPs, abortP]);
       }
-      await Promise.any([...runningPs, abortP]);
+    } finally {
+      // Clean up abort promise when done
+      abortP.cancel();
+      await abortP;
     }
   }
 }

src/nodes/NodeManager.ts

@@ -281,8 +281,14 @@ class NodeManager {
       const task = await this.updateRefreshBucketDelay(i, 0, false);
       refreshBuckets.push(task.promise());
     }
-    const signalProm = utils.signalPromise(ctx.signal).catch(() => {});
-    await Promise.race([Promise.all(refreshBuckets), signalProm]);
+    const signalProm = utils.signalPromise(ctx.signal);
+    await Promise.race([Promise.all(refreshBuckets), signalProm]).finally(
+      async () => {
+        // Clean up signal promise when done
+        signalProm.cancel();
+        await signalProm;
+      },
+    );
   };
 
   public readonly syncNodeGraphHandlerId: TaskHandlerId =
@@ -652,6 +658,7 @@ class NodeManager {
   ): Promise<[[Host, Port], NodeContactAddressData]> {
     // Setting up intermediate signal
     const abortController = new AbortController();
+    utils.setMaxListeners(abortController.signal);
     const newCtx = {
       timer: ctx.timer,
       signal: abortController.signal,
@@ -787,6 +794,7 @@ class NodeManager {
   ): Promise<[[Host, Port], NodeContactAddressData]> {
     // Setting up intermediate signal
     const abortController = new AbortController();
+    utils.setMaxListeners(abortController.signal);
     const newCtx = {
       timer: ctx.timer,
       signal: abortController.signal,

src/utils/utils.ts

@@ -8,7 +8,9 @@ import type {
 import os from 'os';
 import process from 'process';
 import path from 'path';
+import nodesEvents from 'events';
 import lexi from 'lexicographic-integer';
+import { PromiseCancellable } from '@matrixai/async-cancellable';
 import * as utilsErrors from './errors';
 
 const AsyncFunction = (async () => {}).constructor;
@@ -237,19 +239,20 @@ function promise<T = void>(): PromiseDeconstructed<T> {
  * Promise constructed from signal
  * This rejects when the signal is aborted
  */
-function signalPromise(signal: AbortSignal): Promise<void> {
-  return new Promise<void>((_, reject) => {
-    if (signal.aborted) {
-      reject(signal.reason);
-      return;
-    }
-    signal.addEventListener(
-      'abort',
-      () => {
-        reject(signal.reason);
-      },
-      { once: true },
-    );
+//  fixme: There is also a one signal to many `signalPromise` relationship in the NM connection queue that needs to be fixed.
+function signalPromise(signal: AbortSignal): PromiseCancellable<void> {
+  return new PromiseCancellable((resolve, _, signalCancel) => {
+    // Short circuit if signal already aborted
+    if (signal.aborted) return resolve();
+    // Short circuit if promise is already cancelled
+    if (signalCancel.aborted) return resolve();
+    const handler = () => {
+      signalCancel.removeEventListener('abort', handler);
+      signal.removeEventListener('abort', handler);
+      resolve();
+    };
+    signal.addEventListener('abort', handler, { once: true });
+    signalCancel.addEventListener('abort', handler, { once: true });
   });
 }
 
@@ -465,6 +468,20 @@ async function yieldMicro() {
   return await new Promise<void>((r) => queueMicrotask(r));
 }
 
+/**
+ * Increases the total number of registered event handlers before a node warning is emitted.
+ * In most cases this is not needed but in the case where you have one event emitter for multiple handlers you'll need
+ * to increase the limit.
+ * @param target - The specific `EventTarget` or `EventEmitter` to increase the warning for.
+ * @param limit - The limit before the warning is emitted, defaults to 100000.
+ */
+function setMaxListeners(
+  target: EventTarget | NodeJS.EventEmitter,
+  limit: number = 100000,
+) {
+  nodesEvents.setMaxListeners(limit, target);
+}
+
 export {
   AsyncFunction,
   GeneratorFunction,
@@ -503,4 +520,5 @@ export {
   bufferWrap,
   isBufferSource,
   yieldMicro,
+  setMaxListeners,
 };