diff --git a/packages/requester-node-http/src/__tests__/node-http-requester.test.ts b/packages/requester-node-http/src/__tests__/node-http-requester.test.ts index 123cfcb81..9c30b1f1a 100644 --- a/packages/requester-node-http/src/__tests__/node-http-requester.test.ts +++ b/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) => { + 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((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((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((resolve) => { + server.listen(0, '127.0.0.1', resolve); + }); + const { port } = server.address() as AddressInfo; + await new Promise((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); + }); +}); diff --git a/packages/requester-node-http/src/createHttpRequester.ts b/packages/requester-node-http/src/createHttpRequester.ts index 6791114ed..c1fc9be97 100644 --- a/packages/requester-node-http/src/createHttpRequester.ts +++ b/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 = { @@ -76,7 +91,8 @@ 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) => { @@ -84,31 +100,35 @@ export function createHttpRequester({ }); 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 }); });