Assert bodies consumed

This change throws an uncaught exception if we forget to consume a
response body in a test. Unconsumed bodies may cause `undici` to hang
or throw socket errors.

Author: mrbbot

ava.config.mjs

@@ -16,4 +16,7 @@ export default {
     compile: false,
     rewritePaths,
   },
+  environmentVariables: {
+    MINIFLARE_ASSERT_BODIES_CONSUMED: "true",
+  },
 };

packages/miniflare/src/index.ts

@@ -1091,6 +1091,25 @@ export class Miniflare {
       throw reviveError(this.#workerSrcOpts, caught);
     }
 
+    if (
+      process.env.MINIFLARE_ASSERT_BODIES_CONSUMED !== undefined &&
+      response.body !== null
+    ) {
+      // Throw an uncaught exception if the body from this response isn't
+      // consumed "immediately". `undici` may hang or throw socket errors if we
+      // don't remember to do this:
+      // https://github.com/nodejs/undici/issues/583#issuecomment-1577468249
+      const originalLimit = Error.stackTraceLimit;
+      Error.stackTraceLimit = Infinity;
+      const error = new Error(
+        "`body` returned from `Miniflare#dispatchFetch()` not consumed immediately"
+      );
+      Error.stackTraceLimit = originalLimit;
+      setImmediate(() => {
+        if (!response.bodyUsed) throw error;
+      });
+    }
+
     return response;
   };
 
@@ -1109,7 +1128,8 @@ export class Miniflare {
   async _getProxyClient(): Promise<ProxyClient> {
     this.#checkDisposed();
     await this.ready;
-    return this.#proxyClient!;
+    assert(this.#proxyClient !== undefined);
+    return this.#proxyClient;
   }
 
   async getBindings<Env = Record<string, unknown>>(

packages/miniflare/src/plugins/core/proxy/client.ts

@@ -285,6 +285,7 @@ class ProxyStubHandler<T extends object> implements ProxyHandler<T> {
       // Need to `.pipeThrough()` here otherwise we'll get
       // `TypeError: Response body object should not be disturbed or locked`
       // when trying to construct a `Response` with the stream.
+      // TODO(soon): add support for MINIFLARE_ASSERT_BODIES_CONSUMED here
       unbufferedStream = rest.pipeThrough(new TransformStream());
     }
 

packages/miniflare/src/plugins/core/proxy/fetch-sync.ts

@@ -60,6 +60,7 @@ port.addEventListener("message", async (event) => {
     try {
       port.postMessage({ id, error });
     } catch {
+      // If error failed to serialise, post simplified version
       port.postMessage({ id, error: new Error(String(error)) });
     }
   }
@@ -130,6 +131,7 @@ export class SynchronousFetcher {
         // synchronously fetch from internal Miniflare code (e.g. proxy server)
         throw reviveError([], caught);
       }
+      // TODO(soon): add support for MINIFLARE_ASSERT_BODIES_CONSUMED here
       return { status, headers, body };
     } else {
       throw message.error;

packages/miniflare/test/index.spec.ts

@@ -503,6 +503,7 @@ test("Miniflare: HTTPS fetches using browser CA certificates", async (t) => {
   t.teardown(() => mf.dispose());
   const res = await mf.dispatchFetch("http://localhost");
   t.true(res.ok);
+  await res.arrayBuffer(); // (drain)
 });
 
 test("Miniflare: accepts https requests", async (t) => {
@@ -522,6 +523,7 @@ test("Miniflare: accepts https requests", async (t) => {
 
   const res = await mf.dispatchFetch("https://localhost");
   t.true(res.ok);
+  await res.arrayBuffer(); // (drain)
 
   t.assert(log.logs[0][1].startsWith("Ready on https://"));
 });

packages/miniflare/test/plugins/queues/index.spec.ts

@@ -41,7 +41,7 @@ test("flushes partial and full batches", async (t) => {
             } else if (url.pathname === "/batch") {
               await env.QUEUE.sendBatch(body);
             }
-            return new Response();
+            return new Response(null, { status: 204 });
           }
         }`,
       },
@@ -232,7 +232,7 @@ test("sends all structured cloneable types", async (t) => {
             await env.QUEUE.sendBatch(Object.entries(VALUES).map(
               ([key, value]) => ({ body: { name: key, value } })
             ));
-            return new Response();
+            return new Response(null, { status: 204 });
           },
           async queue(batch, env, ctx) {
             let error;
@@ -305,7 +305,7 @@ test("retries messages", async (t) => {
         const url = new URL(request.url);
         const body = await request.json();
         await env.QUEUE.sendBatch(body);
-        return new Response();
+        return new Response(null, { status: 204 });
       },
       async queue(batch, env, ctx) {
         const res = await env.RETRY_FILTER.fetch("http://localhost", {
@@ -532,7 +532,7 @@ test("moves to dead letter queue", async (t) => {
         const url = new URL(request.url);
         const body = await request.json();
         await env.BAD_QUEUE.sendBatch(body);
-        return new Response();
+        return new Response(null, { status: 204 });
       },
       async queue(batch, env, ctx) {
         const res = await env.RETRY_FILTER.fetch("http://localhost", {
@@ -638,7 +638,7 @@ test("operations permit strange queue names", async (t) => {
       async fetch(request, env, ctx) {
         await env.QUEUE.send("msg1");
         await env.QUEUE.sendBatch([{ body: "msg2" }]);
-        return new Response();
+        return new Response(null, { status: 204 });
       },
       async queue(batch, env, ctx) {
         await env.REPORTER.fetch("http://localhost", {
@@ -718,7 +718,8 @@ test("supports message contentTypes", async (t) => {
   },
 };`,
   });
-  await mf.dispatchFetch("http://localhost");
+  const res = await mf.dispatchFetch("http://localhost");
+  await res.arrayBuffer(); // (drain)
   timers.timestamp += 1000;
   await timers.waitForTasks();
   const batch = await promise;

types/env.d.ts

@@ -2,5 +2,6 @@ declare namespace NodeJS {
   export interface ProcessEnv {
     NODE_ENV?: string;
     NODE_EXTRA_CA_CERTS?: string;
+    MINIFLARE_ASSERT_BODIES_CONSUMED?: string;
   }
 }