Prevent abort listener leaks in async client requests

lib/client.js

@@ -136,60 +136,61 @@ class Client extends Entity {
       let sequenceNumber = null;
       let isResolved = false;
       let isTimeout = false;
+      let effectiveSignal = options.signal;
+      let timeoutSignal;
+
+      const onTimeout = () => {
+        isTimeout = true;
+      };
 
       const cleanup = () => {
         if (sequenceNumber !== null) {
           this._sequenceNumberToCallbackMap.delete(sequenceNumber);
         }
+        if (effectiveSignal) {
+          effectiveSignal.removeEventListener('abort', onAbort);
+        }
+        if (timeoutSignal) {
+          timeoutSignal.removeEventListener('abort', onTimeout);
+        }
         isResolved = true;
       };
 
-      let effectiveSignal = options.signal;
+      const onAbort = () => {
+        if (!isResolved) {
+          cleanup();
+          const error = isTimeout
+            ? new TimeoutError('Service request', options.timeout, {
+                entityType: 'service',
+                entityName: this._serviceName,
+              })
+            : new AbortError('Service request', undefined, {
+                entityType: 'service',
+                entityName: this._serviceName,
+              });
+          reject(error);
+        }
+      };
 
       if (options.timeout !== undefined && options.timeout >= 0) {
-        const timeoutSignal = AbortSignal.timeout(options.timeout);
-
-        timeoutSignal.addEventListener('abort', () => {
-          isTimeout = true;
-        });
+        timeoutSignal = AbortSignal.timeout(options.timeout);
 
         if (options.signal) {
           effectiveSignal = AbortSignal.any([options.signal, timeoutSignal]);
         } else {
           effectiveSignal = timeoutSignal;
         }
+
+        timeoutSignal.addEventListener('abort', onTimeout, { once: true });
       }
 
       if (effectiveSignal) {
         if (effectiveSignal.aborted) {
-          const error = isTimeout
-            ? new TimeoutError('Service request', options.timeout, {
-                entityType: 'service',
-                entityName: this._serviceName,
-              })
-            : new AbortError('Service request', undefined, {
-                entityType: 'service',
-                entityName: this._serviceName,
-              });
-          reject(error);
+          onAbort();
           return;
         }
 
-        effectiveSignal.addEventListener('abort', () => {
-          if (!isResolved) {
-            cleanup();
-            const error = isTimeout
-              ? new TimeoutError('Service request', options.timeout, {
-                  entityType: 'service',
-                  entityName: this._serviceName,
-                })
-              : new AbortError('Service request', undefined, {
-                  entityType: 'service',
-                  entityName: this._serviceName,
-                });
-            reject(error);
-          }
-        });
+        effectiveSignal.addEventListener('abort', onAbort, { once: true });
       }
 
       try {

test/test-client.js

@@ -1,4 +1,5 @@
 import assert from 'assert';
+import { getEventListeners } from 'events';
 import sinon from 'sinon';
 import rclnodejsBinding from '../lib/native_loader.js';
 import Client from '../lib/client.js';
@@ -191,6 +192,103 @@ describe('Client coverage testing', function () {
     assert.deepStrictEqual(result, new MockTypeClass.Response());
   });
 
+  it('sendRequestAsync releases abort listeners after successful requests', async function () {
+    const client = new Client(
+      mockHandle,
+      mockNodeHandle,
+      'test_service',
+      MockTypeClass,
+      {}
+    );
+    const controller = new AbortController();
+    const callerListener = sinon.spy();
+    controller.signal.addEventListener('abort', callerListener);
+
+    for (let requestIndex = 0; requestIndex < 3; requestIndex++) {
+      const promise = client.sendRequestAsync(
+        { a: requestIndex },
+        { signal: controller.signal }
+      );
+      client.processResponse(12345, new MockTypeClass.Response());
+      assert.deepStrictEqual(await promise, { sum: 3 });
+      assert.strictEqual(client._sequenceNumberToCallbackMap.size, 0);
+      assert.deepStrictEqual(getEventListeners(controller.signal, 'abort'), [
+        callerListener,
+      ]);
+    }
+
+    controller.abort();
+    assert.ok(callerListener.calledOnce);
+  });
+
+  for (const outcome of [
+    'success',
+    'manual abort',
+    'timeout',
+    'serialization error',
+    'send error',
+    'pre-aborted signal',
+  ]) {
+    it(`sendRequestAsync cleans up timeout and abort listeners on ${outcome}`, async function () {
+      const client = new Client(
+        mockHandle,
+        mockNodeHandle,
+        'test_service',
+        MockTypeClass,
+        {}
+      );
+      const controller = new AbortController();
+      const timeoutController = new AbortController();
+      const timeoutStub = sandbox
+        .stub(AbortSignal, 'timeout')
+        .returns(timeoutController.signal);
+      const combinedSignalSpy = sandbox.spy(AbortSignal, 'any');
+      const failure = new Error('Request failed');
+
+      if (outcome === 'pre-aborted signal') {
+        controller.abort();
+      } else if (outcome === 'serialization error') {
+        sandbox
+          .stub(MockTypeClass.Request.prototype, 'serialize')
+          .throws(failure);
+      } else if (outcome === 'send error') {
+        rclnodejsBinding.sendRequest.throws(failure);
+      }
+
+      const promise = client.sendRequestAsync(
+        { a: 1 },
+        { signal: controller.signal, timeout: 1000 }
+      );
+      const effectiveSignal = combinedSignalSpy.firstCall.returnValue;
+
+      if (outcome === 'success') {
+        client.processResponse(12345, new MockTypeClass.Response());
+        assert.deepStrictEqual(await promise, { sum: 3 });
+      } else {
+        const expectedError = outcome.endsWith('error')
+          ? failure
+          : { name: outcome === 'timeout' ? 'TimeoutError' : 'AbortError' };
+        const rejected = assert.rejects(promise, expectedError);
+        if (outcome === 'manual abort') controller.abort();
+        if (outcome === 'timeout') timeoutController.abort();
+        await rejected;
+      }
+
+      assert.ok(timeoutStub.calledOnceWithExactly(1000));
+      assert.strictEqual(client._sequenceNumberToCallbackMap.size, 0);
+      for (const signal of [
+        controller.signal,
+        timeoutController.signal,
+        effectiveSignal,
+      ]) {
+        assert.deepStrictEqual(getEventListeners(signal, 'abort'), []);
+      }
+      if (outcome === 'pre-aborted signal') {
+        assert.ok(rclnodejsBinding.sendRequest.notCalled);
+      }
+    });
+  }
+
   it('sendRequestAsync handles timeout', async function () {
     const client = new Client(
       mockHandle,