Workaround "Error: Network connection lost." in POSTs from workerd to workerd (#5106)

* ensure request.body is drained

potential workaround for #5095

* regression test

* fixup! regression test

* fixup! regression test

Use larger body and ensure request read before response sent

* allow request draining to be disabled and add changeset

* remove unnecessary export

* fixup! ensure request.body is drained

---------

Co-authored-by: Peter Bacon Darwin <[email protected]>
Co-authored-by: bcoll <[email protected]>

Author: 3 people

.changeset/empty-zoos-beam.md

@@ -0,0 +1,7 @@
+---
+"wrangler": patch
+---
+
+fix: automatically drain incoming request bodies
+
+Previously, requests sent to `wrangler dev` with unconsumed bodies could result in `Network connection lost` errors. This change attempts to work around the issue by ensuring incoming request bodies are drained if they're not used. This is a temporary fix whilst we try to address the underlying issue. Whilst we don't think this change will introduce any other issues, it can be disabled by setting the `WRANGLER_DISABLE_REQUEST_BODY_DRAINING=true` environment variable. Note this fix is only applied if you've enabled Wrangler's bundling—`--no-bundle` mode continues to have the previous behaviour.

fixtures/worker-app/tests/undrained-body.test.ts

@@ -0,0 +1,44 @@
+import { resolve } from "path";
+import { fetch } from "undici";
+import { afterAll, beforeAll, describe, it } from "vitest";
+import { runWranglerDev } from "../../shared/src/run-wrangler-long-lived";
+
+describe("wrangler dev", () => {
+	let ip: string, port: number, stop: (() => Promise<unknown>) | undefined;
+
+	beforeAll(async () => {
+		({ ip, port, stop } = await runWranglerDev(resolve(__dirname, ".."), [
+			"--port=0",
+			"--inspector-port=0",
+		]));
+	});
+
+	afterAll(async () => {
+		await stop?.();
+	});
+
+	// https://github.com/cloudflare/workers-sdk/issues/5095
+	it("should not fail requests if the Worker does not drain the body", async ({
+		expect,
+	}) => {
+		const COUNT = 30;
+		const requests: boolean[] = [];
+		const errors: string[] = [];
+
+		const body = new Uint8Array(2_000);
+		for (let i = 0; i < COUNT; i++) {
+			const response = await fetch(`http://${ip}:${port}/random`, {
+				method: "POST",
+				body,
+			});
+			requests.push(response.ok);
+			if (!response.ok) {
+				errors.push(await response.text());
+			}
+		}
+
+		expect(requests.length).toBe(COUNT);
+		expect(errors).toEqual([]);
+		expect(requests).toEqual(Array.from({ length: COUNT }).map((i) => true));
+	});
+});

packages/wrangler/src/deployment-bundle/bundle.ts

@@ -138,6 +138,17 @@ export async function bundleWorker(
 	// At this point, we take the opportunity to "wrap" the worker with middleware.
 	const middlewareToLoad: MiddlewareLoader[] = [];
 
+	if (
+		targetConsumer === "dev" &&
+		!process.env.WRANGLER_DISABLE_REQUEST_BODY_DRAINING
+	) {
+		middlewareToLoad.push({
+			name: "ensure-req-body-drained",
+			path: "templates/middleware/middleware-ensure-req-body-drained.ts",
+			supports: ["modules", "service-worker"],
+		});
+	}
+
 	if (targetConsumer === "dev" && !!testScheduled) {
 		middlewareToLoad.push({
 			name: "scheduled",

packages/wrangler/templates/middleware/middleware-ensure-req-body-drained.ts

@@ -0,0 +1,18 @@
+import type { Middleware } from "./common";
+
+const drainBody: Middleware = async (request, env, _ctx, middlewareCtx) => {
+	try {
+		return await middlewareCtx.next(request, env);
+	} finally {
+		try {
+			if (request.body !== null && !request.bodyUsed) {
+				const reader = request.body.getReader();
+				while (!(await reader.read()).done) {}
+			}
+		} catch (e) {
+			console.error("Failed to drain the unused request body.", e);
+		}
+	}
+};
+
+export default drainBody;

packages/wrangler/turbo.json

@@ -39,7 +39,8 @@
 				"HTTP_PROXY",
 				"CI_OS",
 				"SENTRY_DSN",
-				"SYSTEMROOT"
+				"SYSTEMROOT",
+				"WRANGLER_DISABLE_REQUEST_BODY_DRAINING"
 			]
 		}
 	}