fix(cli): forward function requests without cloning (CLI-2415)

apps/cli/src/shared/functions/serve-main-offline.e2e.test.ts

@@ -78,17 +78,22 @@ const KONG_FUNCTIONS_CONFIG = JSON.stringify({
 });
 const CUSTOM_FUNCTION = `import { sharedValue } from "../_shared/value.ts";
 
-Deno.serve(() => new Response("ok", {
-  headers: {
-    "X-Custom-Id": "abc123",
-    "X-Function-Slug": Deno.env.get("SUPABASE_FUNCTION_SLUG") ?? "",
-    "X-Shared-Import": sharedValue,
-    "X-Shared": Deno.env.get("SHARED") ?? "",
-    "X-Function-Only": Deno.env.get("FUNCTION_ONLY") ?? "",
-    "X-Global-Only": Deno.env.get("GLOBAL_ONLY") ?? "",
-    "Access-Control-Expose-Headers": "X-Custom-Id",
-  },
-}));`;
+Deno.serve((req) => {
+  if (req.headers.get("x-reject-before-body") === "true") {
+    return new Response("rejected", { status: 400 });
+  }
+  return new Response("ok", {
+    headers: {
+      "X-Custom-Id": "abc123",
+      "X-Function-Slug": Deno.env.get("SUPABASE_FUNCTION_SLUG") ?? "",
+      "X-Shared-Import": sharedValue,
+      "X-Shared": Deno.env.get("SHARED") ?? "",
+      "X-Function-Only": Deno.env.get("FUNCTION_ONLY") ?? "",
+      "X-Global-Only": Deno.env.get("GLOBAL_ONLY") ?? "",
+      "Access-Control-Expose-Headers": "X-Custom-Id",
+    },
+  });
+});`;
 const NESTED_FUNCTION = `Deno.serve(() => new Response("ok", {
   headers: {
     "X-Function-Slug": Deno.env.get("SUPABASE_FUNCTION_SLUG") ?? "",
@@ -144,6 +149,21 @@ function containerLogs(container: string): string {
   return `${result.stdout ?? ""}\n${result.stderr ?? ""}`;
 }
 
+async function fetchFunctionWithDiagnostics(
+  url: string,
+  diagnosticContainers: readonly string[],
+  init: RequestInit,
+): Promise<Response> {
+  try {
+    return await fetch(url, init);
+  } catch (cause) {
+    const diagnostics = diagnosticContainers
+      .map((container) => `${container} logs:\n${containerLogs(container)}`)
+      .join("\n");
+    throw new Error(`Function request to ${url} failed.\n${diagnostics}`, { cause });
+  }
+}
+
 async function fetchColdFunction(
   url: string,
   diagnosticContainers: readonly string[],
@@ -340,7 +360,7 @@ describe("functions serve runtime template (offline)", () => {
   );
 
   test.skipIf(!dockerAvailable)(
-    "preserves function env and CORS headers and exposes JWT errors through Kong",
+    "preserves function env and CORS headers, exposes JWT errors, and returns early responses through Kong",
     { timeout: SERVE_OFFLINE_TEST_TIMEOUT_MS },
     async () => {
       const imageDeadline = resolveDeadline();
@@ -492,6 +512,18 @@ describe("functions serve runtime template (offline)", () => {
         expect(aliasResponse.headers.get("x-shared-import")).toBe("shared-import-ok");
         expect(nestedResponse.status).toBe(200);
         expect(nestedResponse.headers.get("x-function-slug")).toBe("nested-worker-path");
+        const earlyResponse = await fetchFunctionWithDiagnostics(
+          `${functionsUrl}/custom`,
+          diagnosticContainers,
+          {
+            method: "POST",
+            headers: { "x-reject-before-body": "true" },
+            body: new Uint8Array(128 * 1024),
+            signal: AbortSignal.timeout(5_000),
+          },
+        );
+        expect(earlyResponse.status).toBe(400);
+        expect(await earlyResponse.text()).toBe("rejected");
         const runtimeLogs = containerLogs(runtimeContainer);
         expect(runtimeLogs).toContain("Functions config:");
         expect(runtimeLogs).toContain('"custom"');

apps/cli/src/shared/functions/serve.main.ts

@@ -288,12 +288,13 @@ export function prepareUserRequest(req: Request): Request {
   const clonedURL = new URL(req.url);
   const forwardedHost = req.headers.get("x-forwarded-host");
   clonedURL.hostname = forwardedHost ?? clonedURL.hostname;
-  const clonedReq = new Request(clonedURL, req.clone());
+  // Cloning tees the body, so an unread branch can stall early worker responses.
+  const forwardedReq = new Request(clonedURL, req);
 
-  clonedReq.headers.delete("sb-api-key");
-  EdgeRuntime.applySupabaseTag(req, clonedReq);
+  forwardedReq.headers.delete("sb-api-key");
+  EdgeRuntime.applySupabaseTag(req, forwardedReq);
 
-  return clonedReq;
+  return forwardedReq;
 }
 
 Deno.serve({

packages/stack/src/functions/serve.main.ts

@@ -216,10 +216,11 @@ export function prepareUserRequest(request: Request): Request {
   const url = new URL(request.url);
   const forwardedHost = request.headers.get("x-forwarded-host");
   if (forwardedHost) url.hostname = forwardedHost;
-  const cloned = new Request(url, request.clone());
-  cloned.headers.delete("sb-api-key");
-  EdgeRuntime.applySupabaseTag(request, cloned);
-  return cloned;
+  // Cloning tees the body, so an unread branch can stall early worker responses.
+  const forwarded = new Request(url, request);
+  forwarded.headers.delete("sb-api-key");
+  EdgeRuntime.applySupabaseTag(request, forwarded);
+  return forwarded;
 }
 
 Deno.serve({

packages/stack/src/public/whole-stack.e2e.test.ts

@@ -412,7 +412,12 @@ const activate = async <A>(
   return result;
 };
 
-const request = (base: string, path: string, init: RequestInit = {}): Promise<Response> => {
+const request = (
+  base: string,
+  path: string,
+  init: RequestInit = {},
+  options: Readonly<{ expectedStatus?: number }> = {},
+): Promise<Response> => {
   const url = new URL(path, `${base.replace(/\/$/u, "")}/`);
   const program = Effect.gen(function* () {
     const webRequest = new Request(url.href, init);
@@ -430,7 +435,11 @@ const request = (base: string, path: string, init: RequestInit = {}): Promise<Re
     }
     const response = yield* HttpClient.execute(outgoing);
     const body = yield* response.arrayBuffer;
-    if (response.status < 200 || response.status >= 300)
+    const statusMatches =
+      options.expectedStatus === undefined
+        ? response.status >= 200 && response.status < 300
+        : response.status === options.expectedStatus;
+    if (!statusMatches)
       return yield* new E2ERequestError({
         message: `${init.method ?? "GET"} ${url} returned ${response.status}: ${new TextDecoder().decode(body)}`,
       });
@@ -610,7 +619,10 @@ const serviceHeaders = (credentials: PromiseStackCredentials): Record<string, st
   apiHeaders(credentials, credentials.api.serviceRoleJwt);
 
 const functionSource = (table: string, marker: string): string => `
-Deno.serve(async () => {
+Deno.serve(async (request) => {
+  if (request.headers.get("x-reject-before-body") === "true") {
+    return new Response("rejected", { status: 400 });
+  }
   console.log("${marker}");
   let publishableKey: unknown;
   try {
@@ -1024,6 +1036,20 @@ const exerciseWholeStackFunctions = async (scenario: WholeStackScenario): Promis
         rows: expect.arrayContaining([{ id: 1, payload: markers.first }]),
       }),
     );
+    const earlyResponse = await request(
+      api.url,
+      functionPath,
+      {
+        method: "POST",
+        headers: { ...apiHeaders(credentials), "x-reject-before-body": "true" },
+        body: new Uint8Array(128 * 1024),
+        signal: AbortSignal.timeout(5_000),
+      },
+      { expectedStatus: 400 },
+    );
+    expect(earlyResponse.status).toBe(400);
+    expect(await earlyResponse.text()).toBe("rejected");
+
     await writeFile(
       join(projectRoot, "supabase", "functions", functionSlug, "index.ts"),
       functionSource(table, markers.second),