diff --git a/.changeset/fuzzy-pears-repair.md b/.changeset/fuzzy-pears-repair.md new file mode 100644 index 00000000000..abace30d820 --- /dev/null +++ b/.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. diff --git a/packages/vite-plugin-cloudflare/src/__tests__/utils.spec.ts b/packages/vite-plugin-cloudflare/src/__tests__/utils.spec.ts index 846cbfa37da..cfe6729887e 100644 --- a/packages/vite-plugin-cloudflare/src/__tests__/utils.spec.ts +++ b/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[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((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[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((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((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(); + } + }); }); diff --git a/packages/vite-plugin-cloudflare/src/utils.ts b/packages/vite-plugin-cloudflare/src/utils.ts index 8609cfe6b80..f30ca3dc063 100644 --- a/packages/vite-plugin-cloudflare/src/utils.ts +++ b/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 { + let cleanup: (() => void) | undefined; + + return new ReadableStream({ + 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");