Handle backpressure correctly in response stream

Fixes #76

Author: mjackson

packages/node-fetch-server/CHANGELOG.md

@@ -2,6 +2,10 @@
 
 This is the changelog for [`node-fetch-server`](https://github.com/mjackson/remix-the-web/tree/main/packages/node-fetch-server). It follows [semantic versioning](https://semver.org/).
 
+## HEAD
+
+- Handle backpressure correctly in response streaming
+
 ## v0.7.0 (2025-06-06)
 
 - Add `/src` to npm package, so "go to definition" goes to the actual source

packages/node-fetch-server/src/lib/request-listener.test.ts

@@ -205,6 +205,74 @@ describe('createRequestListener', () => {
       listener(req, res);
     });
   });
+
+  it('handles backpressure when writing response chunks', async () => {
+    await new Promise<void>((resolve) => {
+      let handler: FetchHandler = async () => {
+        // Create a response with multiple chunks
+        let chunks = ['chunk1', 'chunk2', 'chunk3', 'chunk4', 'chunk5'];
+        let body = new ReadableStream({
+          async start(controller) {
+            for (let chunk of chunks) {
+              controller.enqueue(new TextEncoder().encode(chunk));
+            }
+            controller.close();
+          },
+        });
+        return new Response(body);
+      };
+
+      let listener = createRequestListener(handler);
+      assert.ok(listener);
+
+      let req = createMockRequest();
+
+      // Create a custom mock response that properly extends EventEmitter
+      let writtenChunks: Uint8Array[] = [];
+      let writeCallCount = 0;
+      let drainListenerCalled = false;
+
+      let writable = new stream.Writable();
+      let res = Object.assign(writable, {
+        req,
+        writeHead() {},
+        write(chunk: Uint8Array) {
+          writtenChunks.push(chunk);
+          writeCallCount++;
+
+          // Simulate backpressure on chunks 2 and 4
+          if (writeCallCount === 2 || writeCallCount === 4) {
+            setTimeout(() => {
+              writable.emit('drain');
+            }, 0);
+            return false; // Indicate buffer is full
+          }
+          return true; // Indicate buffer has space
+        },
+        end() {
+          assert.equal(writtenChunks.length, 5);
+          assert.equal(writeCallCount, 5);
+
+          // Verify backpressure was handled
+          assert.ok(drainListenerCalled, 'drain listener should have been registered');
+
+          // Verify chunk contents
+          let receivedText = writtenChunks.map((chunk) => new TextDecoder().decode(chunk)).join('');
+          assert.equal(receivedText, 'chunk1chunk2chunk3chunk4chunk5');
+
+          resolve();
+        },
+        once(event: string, callback: () => void) {
+          if (event === 'drain') {
+            drainListenerCalled = true;
+          }
+          stream.Writable.prototype.once.call(writable, event, callback);
+        },
+      }) as unknown as http.ServerResponse;
+
+      listener(req, res);
+    });
+  });
 });
 
 function createMockRequest({

packages/node-fetch-server/src/lib/request-listener.ts

@@ -217,7 +217,11 @@ export async function sendResponse(
   if (response.body != null && res.req.method !== 'HEAD') {
     for await (let chunk of readStream(response.body)) {
       // @ts-expect-error - Node typings for http2 require a 2nd parameter to write but it's optional
-      res.write(chunk);
+      if (res.write(chunk) === false) {
+        await new Promise<void>((resolve) => {
+          res.once('drain', resolve);
+        });
+      }
     }
   }