Test to check if fetchNodeHttp leaks (#1162)

.changeset/tough-rats-argue.md

@@ -0,0 +1,6 @@
+---
+"@whatwg-node/node-fetch": patch
+"@whatwg-node/fetch": patch
+---
+
+Consume the body with PassThrough

.eslintrc.json

@@ -64,7 +64,7 @@
     "website",
     "scripts",
     "e2e",
-    "benchmark"
+    "benchmarks"
   ],
   "globals": {
     "BigInt": true

package.json

@@ -11,6 +11,7 @@
     "ci:lint": "eslint --ext .ts . --output-file eslint_report.json --format json",
     "clean-dist": "rimraf \"dist\" && rimraf \".bob\"",
     "esm:check": "bob check",
+    "jest-with-gc": "node --expose-gc ./node_modules/.bin/jest",
     "lint": "eslint --ext .ts .",
     "postinstall": "patch-package && husky install",
     "prebuild": "yarn clean-dist",

packages/node-fetch/src/fetchCurl.ts

@@ -1,4 +1,4 @@
-import { Readable } from 'node:stream';
+import { PassThrough, Readable } from 'node:stream';
 import { PonyfillRequest } from './Request.js';
 import { PonyfillResponse } from './Response.js';
 import { defaultHeadersSerializer, isNodeReadable } from './utils.js';
@@ -93,6 +93,7 @@ export function fetchCurl<TResponseJSON = any, TRequestJSON = any>(
     curlHandle.once(
       'stream',
       function streamListener(stream: Readable, status: number, headersBuf: Buffer) {
+        const pipedStream = stream.pipe(new PassThrough());
         const headersFlat = headersBuf
           .toString('utf8')
           .split(/\r?\n|\r/g)
@@ -102,7 +103,7 @@ export function fetchCurl<TResponseJSON = any, TRequestJSON = any>(
                 fetchRequest.redirect === 'error' &&
                 (headerFilter.includes('location') || headerFilter.includes('Location'))
               ) {
-                stream.destroy();
+                pipedStream.destroy();
                 reject(new Error('redirect is not allowed'));
               }
               return true;
@@ -112,14 +113,22 @@ export function fetchCurl<TResponseJSON = any, TRequestJSON = any>(
         const headersInit = headersFlat.map(
           headerFlat => headerFlat.split(/:\s(.+)/).slice(0, 2) as [string, string],
         );
-        resolve(
-          new PonyfillResponse(stream, {
-            status,
-            headers: headersInit,
-            url: fetchRequest.url,
-          }),
-        );
-        streamResolved = stream;
+        pipedStream.on('pause', () => {
+          stream.pause();
+        });
+        pipedStream.on('resume', () => {
+          stream.resume();
+        });
+        pipedStream.on('close', () => {
+          stream.destroy();
+        });
+        const ponyfillResponse = new PonyfillResponse(pipedStream, {
+          status,
+          headers: headersInit,
+          url: fetchRequest.url,
+        });
+        resolve(ponyfillResponse);
+        streamResolved = pipedStream;
       },
     );
     curlHandle.perform();

packages/node-fetch/src/fetchNodeHttp.ts

@@ -1,7 +1,8 @@
 import { request as httpRequest } from 'http';
 import { request as httpsRequest } from 'https';
-import { Readable } from 'stream';
+import { PassThrough, Readable } from 'stream';
 import { createBrotliDecompress, createGunzip, createInflate } from 'zlib';
+import { PonyfillAbortError } from './AbortError.js';
 import { PonyfillRequest } from './Request.js';
 import { PonyfillResponse } from './Response.js';
 import { PonyfillURL } from './URL.js';
@@ -78,6 +79,23 @@ export function fetchNodeHttp<TResponseJSON = any, TRequestJSON = any>(
             return;
           }
         }
+        if (responseBody === nodeResponse) {
+          responseBody = nodeResponse.pipe(new PassThrough());
+          responseBody.on('pause', () => {
+            nodeResponse.pause();
+          });
+          responseBody.on('resume', () => {
+            nodeResponse.resume();
+          });
+          responseBody.on('close', () => {
+            nodeResponse.destroy();
+          });
+          fetchRequest['_signal']?.addEventListener('abort', () => {
+            if (!nodeResponse.destroyed) {
+              responseBody.emit('error', new PonyfillAbortError());
+            }
+          });
+        }
         const ponyfillResponse = new PonyfillResponse(responseBody, {
           status: nodeResponse.statusCode,
           statusText: nodeResponse.statusMessage,

packages/node-fetch/tests/cleanup-resources.spec.ts

@@ -0,0 +1,31 @@
+import { Response } from '@whatwg-node/node-fetch';
+import { createServerAdapter } from '@whatwg-node/server';
+import { runTestsForEachFetchImpl } from '../../server/test/test-fetch';
+import { runTestsForEachServerImpl } from '../../server/test/test-server';
+import { fetchPonyfill } from '../src/fetch';
+
+describe('Cleanup Resources', () => {
+  runTestsForEachFetchImpl(() => {
+    describe('internal calls', () => {
+      runTestsForEachServerImpl(testServer => {
+        beforeEach(() => {
+          testServer.addOnceHandler(createServerAdapter(() => Response.json({ test: 'test' })));
+        });
+        it('should free resources when body is not consumed', async () => {
+          const response = await fetchPonyfill(testServer.url);
+          expect(response.status).toBe(200);
+        });
+      });
+    });
+    describe('external calls', () => {
+      it('http - should free resources when body is not consumed', async () => {
+        const response = await fetchPonyfill('http://google.com');
+        expect(response.status).toBe(200);
+      });
+      it('https - should free resources when body is not consumed', async () => {
+        const response = await fetchPonyfill('https://google.com');
+        expect(response.status).toBe(200);
+      });
+    });
+  });
+});

packages/server/src/utils.ts

@@ -303,7 +303,7 @@ export function sendNodeResponse(
   serverResponse: NodeResponse,
   nodeRequest: NodeRequest,
 ) {
-  if (serverResponse.closed || serverResponse.destroyed) {
+  if (serverResponse.closed || serverResponse.destroyed || serverResponse.writableEnded) {
     return;
   }
   if (!fetchResponse) {

packages/server/test/test-fetch.ts

@@ -7,6 +7,9 @@ export function runTestsForEachFetchImpl(callback: (implementationName: string)
   if (libcurl) {
     describe('libcurl', () => {
       callback('libcurl');
+      afterAll(() => {
+        libcurl.Curl.globalCleanup();
+      });
     });
   }
   describe('node-http', () => {
@@ -20,4 +23,7 @@ export function runTestsForEachFetchImpl(callback: (implementationName: string)
     });
     callback('node-http');
   });
+  afterEach(() => {
+    globalThis?.gc?.();
+  });
 }