fix(requester-node-http): remove listeners after request settles

packages/requester-node-http/src/__tests__/node-http-requester.test.ts

@@ -1,8 +1,9 @@
 import http from 'http';
 import https from 'https';
+import type { AddressInfo } from 'net';
 import nock from 'nock';
 import { Readable } from 'stream';
-import { afterAll, beforeAll, describe, expect, test, vi } from 'vitest';
+import { afterAll, afterEach, beforeAll, describe, expect, test, vi } from 'vitest';
 
 import type { EndRequest } from '@algolia/client-common';
 
@@ -29,6 +30,7 @@ describe('api', () => {
     once: vi.fn(),
     write: vi.fn(),
     end: vi.fn(),
+    removeAllListeners: vi.fn(),
   };
 
   beforeAll(() => {
@@ -319,3 +321,89 @@ describe('response stream error handling', () => {
     expect(response.isTimedOut).toBe(false);
   });
 });
+
+describe('listener cleanup', () => {
+  const originalHttpRequest = http.request;
+
+  afterEach(() => {
+    http.request = originalHttpRequest;
+  });
+
+  function captureHttpRequest(): { req?: http.ClientRequest; res?: http.IncomingMessage } {
+    const captured: { req?: http.ClientRequest; res?: http.IncomingMessage } = {};
+
+    http.request = ((...args: Parameters<typeof originalHttpRequest>) => {
+      const req = originalHttpRequest(...args);
+      captured.req = req;
+      req.prependOnceListener('response', (res) => {
+        captured.res = res;
+      });
+      return req;
+    }) as typeof http.request;
+
+    return captured;
+  }
+
+  test('removes request and response listeners after a successful request', async () => {
+    const captured = captureHttpRequest();
+    const server = http.createServer((_req, res) => {
+      res.writeHead(200, { 'content-type': 'text/plain' });
+      res.end('ok');
+    });
+
+    await new Promise<void>((resolve) => {
+      server.listen(0, '127.0.0.1', resolve);
+    });
+    const { port } = server.address() as AddressInfo;
+
+    try {
+      const response = await requester.send({
+        url: `http://127.0.0.1:${port}/`,
+        method: 'GET',
+        headers: {},
+        data: '',
+        connectTimeout: 1000,
+        responseTimeout: 1000,
+      });
+
+      expect(response.status).toBe(200);
+      expect(response.content).toBe('ok');
+      expect(captured.req).toBeDefined();
+      expect(captured.res).toBeDefined();
+      expect(captured.req!.listenerCount('error')).toBe(0);
+      expect(captured.res!.listenerCount('data')).toBe(0);
+      expect(captured.res!.listenerCount('end')).toBe(0);
+      expect(captured.res!.listenerCount('error')).toBe(0);
+    } finally {
+      await new Promise<void>((resolve, reject) => {
+        server.close((err) => (err ? reject(err) : resolve()));
+      });
+    }
+  });
+
+  test('removes request listeners after a network error', async () => {
+    const captured = captureHttpRequest();
+    const server = http.createServer();
+
+    await new Promise<void>((resolve) => {
+      server.listen(0, '127.0.0.1', resolve);
+    });
+    const { port } = server.address() as AddressInfo;
+    await new Promise<void>((resolve, reject) => {
+      server.close((err) => (err ? reject(err) : resolve()));
+    });
+
+    const response = await requester.send({
+      url: `http://127.0.0.1:${port}/`,
+      method: 'GET',
+      headers: {},
+      data: '',
+      connectTimeout: 1000,
+      responseTimeout: 1000,
+    });
+
+    expect(response.status).toBe(0);
+    expect(captured.req).toBeDefined();
+    expect(captured.req!.listenerCount('error')).toBe(0);
+  });
+});

packages/requester-node-http/src/createHttpRequester.ts

@@ -47,6 +47,21 @@ export function createHttpRequester({
     return new Promise((resolve) => {
       let responseTimeout: NodeJS.Timeout | undefined;
       let connectTimeout: NodeJS.Timeout | undefined;
+      let incomingResponse: http.IncomingMessage | undefined;
+      let req: http.ClientRequest;
+      let cleanedUp = false;
+
+      const cleanup = (): void => {
+        if (cleanedUp) {
+          return;
+        }
+        cleanedUp = true;
+        clearTimeout(connectTimeout);
+        clearTimeout(responseTimeout);
+        incomingResponse?.removeAllListeners();
+        req.removeAllListeners();
+      };
+
       const url = new URL(request.url);
       const path = url.search === null ? url.pathname : `${url.pathname}${url.search}`;
       const privateHeaders: Record<string, string> = {
@@ -76,39 +91,44 @@ export function createHttpRequester({
         options.port = url.port;
       }
 
-      const req = (url.protocol === 'https:' ? https : http).request(options, (response) => {
+      req = (url.protocol === 'https:' ? https : http).request(options, (response) => {
+        incomingResponse = response;
         let contentBuffers: Buffer[] = [];
 
         response.on('data', (chunk) => {
           contentBuffers = contentBuffers.concat(chunk);
         });
 
         response.on('end', () => {
-          clearTimeout(connectTimeout as NodeJS.Timeout);
-          clearTimeout(responseTimeout as NodeJS.Timeout);
-
           let buffer = Buffer.concat(contentBuffers);
           if (response.headers['content-encoding'] === 'gzip') {
             buffer = zlib.gunzipSync(buffer);
           }
 
+          const status = response.statusCode || 0;
+          const headers = toResponseHeaders(response.headers);
+          cleanup();
+
           resolve({
-            status: response.statusCode || 0,
+            status,
             content: buffer.toString(),
-            headers: toResponseHeaders(response.headers),
+            headers,
             isTimedOut: false,
           });
         });
 
         response.on('error', (error) => {
-          clearTimeout(connectTimeout as NodeJS.Timeout);
-          clearTimeout(responseTimeout as NodeJS.Timeout);
+          cleanup();
           resolve({ status: 0, content: error.message, isTimedOut: false });
         });
       });
 
       const createTimeout = (timeout: number, content: string): NodeJS.Timeout => {
         return setTimeout(() => {
+          cleanup();
+          req.on('error', () => {
+            // Swallow errors from destroy() after the request has already settled.
+          });
           req.destroy();
 
           resolve({
@@ -122,8 +142,7 @@ export function createHttpRequester({
       connectTimeout = createTimeout(request.connectTimeout, 'Connection timeout');
 
       req.on('error', (error) => {
-        clearTimeout(connectTimeout as NodeJS.Timeout);
-        clearTimeout(responseTimeout!);
+        cleanup();
         resolve({ status: 0, content: error.message, isTimedOut: false });
       });