[vite-plugin-cloudflare] Handle cancelled request bodies

.changeset/fuzzy-pears-repair.md

@@ -0,0 +1,5 @@
+---
+"@cloudflare/vite-plugin": patch
+---
+
+Handle Worker-side request body cancellation without surfacing stream controller errors in local dev.

packages/vite-plugin-cloudflare/src/__tests__/utils.spec.ts

@@ -1,4 +1,5 @@
 import http from "node:http";
+import net from "node:net";
 import * as path from "node:path";
 import { Response as MiniflareResponse } from "miniflare";
 import { afterEach, beforeEach, describe, test } from "vitest";
@@ -188,4 +189,182 @@ describe("createRequestHandler", () => {
 		});
 		expect(capturedUrls[0]).toBe(`http://127.0.0.1:${port}/path`);
 	});
+
+	test("returns the Worker response when the Worker cancels an oversized body", async ({
+		expect,
+	}) => {
+		const handler = createRequestHandler(async (request) => {
+			const reader = request.body?.getReader();
+			let total = 0;
+
+			if (!reader) {
+				return new MiniflareResponse(null, { status: 204 });
+			}
+
+			try {
+				while (true) {
+					const result = await reader.read();
+					if (result.done) {
+						break;
+					}
+					total += result.value.byteLength;
+
+					if (total > 64 * 1024) {
+						await reader.cancel();
+						return MiniflareResponse.json(
+							{ error: "The request body is too large." },
+							{ status: 413 }
+						);
+					}
+				}
+			} finally {
+				reader.releaseLock();
+			}
+
+			return new MiniflareResponse(null, { status: 204 });
+		});
+
+		httpServer = http.createServer((req, res) => {
+			void handler(
+				req as unknown as Parameters<typeof handler>[0],
+				res,
+				(error: unknown) => {
+					res.statusCode = 500;
+					res.setHeader("content-type", "text/plain");
+					res.end(error instanceof Error ? error.message : String(error));
+				}
+			);
+		});
+
+		await new Promise<void>((r) =>
+			httpServer.listen(0, "127.0.0.1", () => {
+				port = (httpServer.address() as AddressInfo).port;
+				r();
+			})
+		);
+
+		for (let i = 0; i < 300; i++) {
+			const response = await fetch(`http://127.0.0.1:${port}/upload`, {
+				method: "POST",
+				body: new ReadableStream({
+					start(controller) {
+						for (let j = 0; j < 8; j++) {
+							controller.enqueue(new Uint8Array(16 * 1024));
+						}
+						controller.close();
+					},
+				}),
+				duplex: "half",
+			} as RequestInit & { duplex: "half" });
+
+			expect(response.status).toBe(413);
+			expect(await response.json()).toEqual({
+				error: "The request body is too large.",
+			});
+		}
+	});
+
+	test("drains cancelled request bodies so keep-alive connections can be reused", async ({
+		expect,
+	}) => {
+		const handler = createRequestHandler(async (request) => {
+			if (request.method === "GET") {
+				return new MiniflareResponse("OK");
+			}
+
+			const reader = request.body?.getReader();
+			let total = 0;
+
+			if (!reader) {
+				return new MiniflareResponse(null, { status: 204 });
+			}
+
+			try {
+				while (true) {
+					const result = await reader.read();
+					if (result.done) {
+						break;
+					}
+					total += result.value.byteLength;
+
+					if (total > 64 * 1024) {
+						await reader.cancel();
+						return new MiniflareResponse("Too large", { status: 413 });
+					}
+				}
+			} finally {
+				reader.releaseLock();
+			}
+
+			return new MiniflareResponse(null, { status: 204 });
+		});
+
+		httpServer = http.createServer((req, res) => {
+			void handler(
+				req as unknown as Parameters<typeof handler>[0],
+				res,
+				(error: unknown) => {
+					res.statusCode = 500;
+					res.setHeader("content-type", "text/plain");
+					res.end(error instanceof Error ? error.message : String(error));
+				}
+			);
+		});
+
+		await new Promise<void>((r) =>
+			httpServer.listen(0, "127.0.0.1", () => {
+				port = (httpServer.address() as AddressInfo).port;
+				r();
+			})
+		);
+
+		const socket = net.connect(port, "127.0.0.1");
+		try {
+			const responses = await new Promise<string>((resolve, reject) => {
+				let data = "";
+				const timeout = setTimeout(() => {
+					reject(new Error(`Timed out waiting for reused connection: ${data}`));
+				}, 2000);
+
+				socket.on("data", (chunk) => {
+					data += chunk.toString("utf8");
+					if (data.includes("HTTP/1.1 413") && data.includes("HTTP/1.1 200")) {
+						clearTimeout(timeout);
+						resolve(data);
+					}
+				});
+				socket.on("error", (error) => {
+					clearTimeout(timeout);
+					reject(error);
+				});
+
+				socket.write(
+					[
+						"POST /upload HTTP/1.1",
+						`Host: 127.0.0.1:${port}`,
+						"Connection: keep-alive",
+						"Content-Length: 131072",
+						"",
+						"",
+					].join("\r\n")
+				);
+				socket.write(Buffer.alloc(131072));
+				socket.write(
+					[
+						"GET /after-cancel HTTP/1.1",
+						`Host: 127.0.0.1:${port}`,
+						"Connection: close",
+						"",
+						"",
+					].join("\r\n")
+				);
+			});
+
+			expect(responses).toContain("HTTP/1.1 413");
+			expect(responses).toContain("HTTP/1.1 200");
+			expect(responses).toContain("OK");
+		} finally {
+			socket.destroy();
+		}
+	});
 });

packages/vite-plugin-cloudflare/src/utils.ts

@@ -1,6 +1,6 @@
 import * as nodePath from "node:path";
 import * as util from "node:util";
-import { createRequest, sendResponse } from "@remix-run/node-fetch-server";
+import { createHeaders, sendResponse } from "@remix-run/node-fetch-server";
 import {
 	CoreHeaders,
 	Request as MiniflareRequest,
@@ -97,7 +97,11 @@ export function createRequestHandler(
 			// If the header is absent or invalid, `createRequest` falls back to the
 			// connection protocol (`req.socket.encrypted`).
 			const protocol = getForwardedProto(req);
-			request = createRequest(req, res, protocol ? { protocol } : undefined);
+			request = createRequestForIncomingMessage(
+				req,
+				res,
+				protocol ? { protocol } : undefined
+			);
 
 			let response = await handler(toMiniflareRequest(request), req);
 
@@ -126,6 +130,88 @@ export function satisfiesMinimumViteVersion(minVersion: string): boolean {
 	return semverGte(viteVersion, minVersion);
 }
 
+function createRequestForIncomingMessage(
+	req: vite.Connect.IncomingMessage,
+	res: http.ServerResponse,
+	options?: { protocol?: "http:" | "https:"; host?: string }
+): Request {
+	const controller = new AbortController();
+	res.on("close", () => {
+		controller.abort();
+	});
+
+	const method = req.method ?? "GET";
+	const headers = createHeaders(req);
+	const protocol =
+		options?.protocol ??
+		("encrypted" in req.socket && req.socket.encrypted ? "https:" : "http:");
+	const host = options?.host ?? headers.get("Host") ?? "localhost";
+	const url = new URL(req.url ?? "/", `${protocol}//${host}`);
+	const init: RequestInit & { duplex?: "half" } = {
+		method,
+		headers,
+		signal: controller.signal,
+	};
+
+	if (method !== "GET" && method !== "HEAD") {
+		init.body = createCancellableRequestBody(req);
+		init.duplex = "half";
+	}
+
+	return new Request(url, init);
+}
+
+function createCancellableRequestBody(
+	req: vite.Connect.IncomingMessage
+): ReadableStream<Uint8Array> {
+	let cleanup: (() => void) | undefined;
+
+	return new ReadableStream<Uint8Array>({
+		start(controller) {
+			const onData = (chunk: Buffer) => {
+				try {
+					controller.enqueue(
+						new Uint8Array(chunk.buffer, chunk.byteOffset, chunk.byteLength)
+					);
+				} catch (error) {
+					if (
+						error instanceof TypeError &&
+						error.message.includes("Controller is already closed")
+					) {
+						cleanup?.();
+						req.resume();
+						return;
+					}
+					throw error;
+				}
+			};
+			const onEnd = () => {
+				cleanup?.();
+				controller.close();
+			};
+			const onError = (error: Error) => {
+				cleanup?.();
+				controller.error(error);
+			};
+
+			cleanup = () => {
+				req.off("data", onData);
+				req.off("end", onEnd);
+				req.off("error", onError);
+				cleanup = undefined;
+			};
+
+			req.on("data", onData);
+			req.on("end", onEnd);
+			req.on("error", onError);
+		},
+		cancel() {
+			cleanup?.();
+			req.resume();
+		},
+	});
+}
+
 function toMiniflareRequest(request: Request): MiniflareRequest {
 	const host = request.headers.get("Host");
 	const xForwardedHost = request.headers.get("X-Forwarded-Host");