fix(pages-router): stream piped API responses with backpressure

packages/vinext/src/server/pages-api-route.ts

@@ -8,6 +8,7 @@ import {
 } from "../utils/query.js";
 import {
   createPagesReqRes,
+  markVinextStreamedApiResponse,
   parsePagesApiBody,
   type PagesRequestQuery,
   type PagesReqResRequest,
@@ -228,7 +229,15 @@ async function _handlePagesApiRoute(options: HandlePagesApiRouteOptions): Promis
     if (!externalResolver && !resWasPiped && !res.headersSent) {
       res.end();
     }
-    return await responsePromise;
+    const response = await responsePromise;
+    // The response resolves on the first write; if `end()` still has not been
+    // called by the time it settles, the body is a live stream (e.g. a piped
+    // proxy upstream). Mark it so buffering adapters forward it as a stream
+    // instead of holding the whole body in memory until the source closes.
+    if (!res.writableEnded) {
+      markVinextStreamedApiResponse(response);
+    }
+    return response;
   } catch (error) {
     if (error instanceof PagesApiBodyParseError) {
       return new Response(error.message, {

packages/vinext/src/server/pages-node-compat.ts

@@ -254,6 +254,7 @@ class PagesResponseStream extends Writable {
   private controller: ReadableStreamDefaultController | null = null;
   private readonly bufferedChunks: Buffer[] = [];
   private streamEnded = false;
+  private pendingWrite: (() => void) | null = null;
 
   constructor(
     private readonly resolveResponse: (value: Response) => void,
@@ -405,7 +406,22 @@ class PagesResponseStream extends Writable {
       this.bufferedChunks.push(buffer);
     }
     this.resolveOnce();
-    callback();
+    // Propagate consumer backpressure to the Node source: while the response
+    // body's queue is full, hold the write callback until the consumer pulls.
+    // A held callback makes `write()` return false, which pauses any piped
+    // source (e.g. a proxied upstream) instead of queueing chunks without
+    // bound. Writes issued by `end(data)` belong to a complete body — finish
+    // them immediately so `finish` is not gated on the first read.
+    if (
+      this.controller &&
+      !this.streamEnded &&
+      !this.writableEnded &&
+      (this.controller.desiredSize ?? 1) <= 0
+    ) {
+      this.pendingWrite = callback;
+    } else {
+      callback();
+    }
   }
 
   override _final(callback: (error?: Error | null) => void): void {
@@ -423,6 +439,9 @@ class PagesResponseStream extends Writable {
 
   override _destroy(error: Error | null, callback: (error?: Error | null) => void): void {
     this.streamEnded = true;
+    // A write parked for backpressure would otherwise never complete; release
+    // it so piped sources unwind instead of waiting on a dead stream.
+    this.releasePendingWrite();
     if (!this.resolved) {
       if (error) {
         this.resolved = true;
@@ -464,6 +483,12 @@ class PagesResponseStream extends Writable {
     this.resHeaders[name.toLowerCase()] = Array.isArray(value) ? value.join(", ") : value;
   }
 
+  private releasePendingWrite(): void {
+    const callback = this.pendingWrite;
+    this.pendingWrite = null;
+    callback?.();
+  }
+
   private resolveOnce(): void {
     if (this.resolved) {
       return;
@@ -497,6 +522,9 @@ class PagesResponseStream extends Writable {
           }
         }
       },
+      pull: () => {
+        this.releasePendingWrite();
+      },
       cancel: (reason) => {
         this.bufferedChunks.length = 0;
         this.destroy(reason instanceof Error ? reason : new Error("Response body cancelled"));
@@ -507,6 +535,26 @@ class PagesResponseStream extends Writable {
   }
 }
 
+type ResponseWithVinextStreamedApiBody = Response & {
+  __vinextStreamedApiResponse?: boolean;
+};
+
+/**
+ * Marks a Pages API Response whose body was still being written when the
+ * handler settled (streaming/piping), so Node adapters forward it as a stream.
+ * Buffering a live stream would defer delivery until the source closes and
+ * hold the whole body in memory. Complete bodies (`res.json`, `res.send`,
+ * `res.end(data)`) are not marked and keep the buffered path, which preserves
+ * Content-Length.
+ */
+export function markVinextStreamedApiResponse(response: Response): void {
+  (response as ResponseWithVinextStreamedApiBody).__vinextStreamedApiResponse = true;
+}
+
+export function isVinextStreamedApiResponse(response: Response): boolean {
+  return (response as ResponseWithVinextStreamedApiBody).__vinextStreamedApiResponse === true;
+}
+
 export function createPagesReqRes(options: CreatePagesReqResOptions): CreatePagesReqResResult {
   const headersObj: Record<string, string> = {};
   for (const [key, value] of options.request.headers) {

packages/vinext/src/server/pages-request-pipeline.ts

@@ -580,12 +580,21 @@ export async function runPagesRequest(
         apiRequest = cloneRequestWithUrl(request, apiRequestUrl.toString());
       }
       const response = await deps.handleApi(apiRequest, apiLookupUrl, deps.ctx ?? null);
+      const merged = mergeHeaders(response, middlewareHeaders, middlewareStatus);
+      // Preserve the streaming marker so the adapter can decide stream-vs-buffer.
+      // mergeHeaders may create a new Response object (losing non-standard
+      // properties), so copy the marker from the original API response.
+      if (merged !== response) {
+        (merged as { __vinextStreamedApiResponse?: boolean }).__vinextStreamedApiResponse = (
+          response as { __vinextStreamedApiResponse?: boolean }
+        ).__vinextStreamedApiResponse;
+      }
       return {
         type: "response",
         // API routes return arbitrary data; default a missing content-type to
         // application/octet-stream (not text/html) to avoid content sniffing.
         defaultContentType: "application/octet-stream",
-        response: mergeHeaders(response, middlewareHeaders, middlewareStatus),
+        response: merged,
       };
     }
     return {

packages/vinext/src/server/prod-server.ts

@@ -437,12 +437,21 @@ function cancelResponseBody(response: Response): void {
 
 type ResponseWithVinextStreamingMetadata = Response & {
   __vinextStreamedHtmlResponse?: boolean;
+  __vinextStreamedApiResponse?: boolean;
 };
 
 function isVinextStreamedHtmlResponse(response: Response): boolean {
   return (response as ResponseWithVinextStreamingMetadata).__vinextStreamedHtmlResponse === true;
 }
 
+// Set by the Pages API bridge (pages-node-compat.ts) when the Response was
+// resolved while the handler was still writing (streaming/piping). Buffering
+// such a body would defer delivery until the source closes and hold the whole
+// stream in memory.
+function isVinextStreamedApiResponse(response: Response): boolean {
+  return (response as ResponseWithVinextStreamingMetadata).__vinextStreamedApiResponse === true;
+}
+
 function logProdServerStarted(host: string, port: number, purpose: ProdServerOptions["purpose"]) {
   const url = `http://${host}:${port}`;
   if (purpose === "prerender") {
@@ -2009,12 +2018,21 @@ async function startPagesRouterServer(options: PagesRouterServerOptions) {
           res.end("Not Found");
           return;
         }
-        const shouldStream = isVinextStreamedHtmlResponse(response);
+        const streamedApi = isVinextStreamedApiResponse(response);
+        const shouldStream = isVinextStreamedHtmlResponse(response) || streamedApi;
         // Passthrough responses (middleware short-circuits, external proxies, redirects)
         // carry no defaultContentType — send them verbatim without injecting a
-        // Content-Type, matching the pre-refactor behavior. Only buffered render/api
-        // responses below apply a Content-Type fallback.
+        // Content-Type, matching the pre-refactor behavior. Buffered render/api
+        // responses below apply a Content-Type fallback; streamed API responses
+        // apply the same fallback here since they skip the buffered path.
         if (shouldStream || !response.body || result.defaultContentType === undefined) {
+          if (
+            streamedApi &&
+            result.defaultContentType !== undefined &&
+            !response.headers.has("content-type")
+          ) {
+            response.headers.set("content-type", result.defaultContentType);
+          }
           await sendWebResponse(response, req, res, compress);
           return;
         }

packages/vinext/src/shims/unified-request-context.ts

@@ -324,6 +324,9 @@ export function closeAfterResponseWithBody(
   (wrapped as { __vinextStreamedHtmlResponse?: boolean }).__vinextStreamedHtmlResponse = (
     response as { __vinextStreamedHtmlResponse?: boolean }
   ).__vinextStreamedHtmlResponse;
+  (wrapped as { __vinextStreamedApiResponse?: boolean }).__vinextStreamedApiResponse = (
+    response as { __vinextStreamedApiResponse?: boolean }
+  ).__vinextStreamedApiResponse;
   return wrapped;
 }
 

tests/pages-api-route.test.ts

@@ -1,13 +1,14 @@
 import { AsyncLocalStorage } from "node:async_hooks";
 import http from "node:http";
 import type { AddressInfo } from "node:net";
-import { PassThrough } from "node:stream";
+import { PassThrough, Readable } from "node:stream";
 import { gzipSync } from "node:zlib";
 import { describe, expect, it, vi } from "vite-plus/test";
 import {
   handlePagesApiRoute,
   type PagesApiRouteMatch,
 } from "../packages/vinext/src/server/pages-api-route.js";
+import { isVinextStreamedApiResponse } from "../packages/vinext/src/server/pages-node-compat.js";
 
 type PagesApiRouteModule = PagesApiRouteMatch["route"]["module"];
 
@@ -940,6 +941,81 @@ describe("pages api route", () => {
     await expect(response.text()).resolves.toBe(body);
   });
 
+  it("pauses a piped source until the response body is consumed", async () => {
+    // Availability: a fast source (e.g. a proxied upstream) piped into `res`
+    // must pause when the client is not reading, instead of queueing the
+    // entire stream in memory. The write callback is held while the body's
+    // queue is full, so `write()` returns false and `pipe()` pauses the source.
+    const totalChunks = 20;
+    const chunk = Buffer.alloc(64 * 1024, "x");
+    let produced = 0;
+    const source = Readable.from(
+      (function* () {
+        for (let i = 0; i < totalChunks; i++) {
+          produced++;
+          yield chunk;
+        }
+      })(),
+    );
+
+    const response = await handlePagesApiRoute({
+      match: createMatch(
+        (_req, res) => {
+          source.pipe(res);
+        },
+        {},
+        { api: { bodyParser: false } },
+      ),
+      request: new Request("https://example.com/api/backpressure"),
+      url: "/api/backpressure",
+    });
+
+    // Let the pipe run as far as it can while nothing reads the body.
+    for (let i = 0; i < 10; i++) {
+      await new Promise((resolve) => setImmediate(resolve));
+    }
+    // Without backpressure the source drains completely here even though no
+    // consumer has read a single byte.
+    expect(produced).toBeLessThan(totalChunks);
+
+    const body = await response.arrayBuffer();
+    expect(body.byteLength).toBe(totalChunks * chunk.length);
+    expect(produced).toBe(totalChunks);
+  });
+
+  it("marks a response as streamed only while its handler is still writing", async () => {
+    // Contract with Node adapters (prod-server): live streams must be
+    // forwarded as streams, while complete bodies stay on the buffered path
+    // that preserves Content-Length.
+    const upstream = new PassThrough();
+    upstream.write("data");
+    const streamed = await handlePagesApiRoute({
+      match: createMatch(
+        (_req, res) => {
+          // Proxy pattern: the upstream stays open past the handler's return.
+          upstream.pipe(res);
+        },
+        {},
+        { api: { bodyParser: false } },
+      ),
+      request: new Request("https://example.com/api/marker-streamed"),
+      url: "/api/marker-streamed",
+    });
+    expect(isVinextStreamedApiResponse(streamed)).toBe(true);
+    upstream.end("-more");
+    await expect(streamed.text()).resolves.toBe("data-more");
+
+    const buffered = await handlePagesApiRoute({
+      match: createMatch((_req, res) => {
+        res.json({ ok: true });
+      }),
+      request: new Request("https://example.com/api/marker-buffered"),
+      url: "/api/marker-buffered",
+    });
+    expect(isVinextStreamedApiResponse(buffered)).toBe(false);
+    await expect(buffered.json()).resolves.toEqual({ ok: true });
+  });
+
   it("streams a piped request body through to the response", async () => {
     const body = "piped-body-content";
     const response = await handlePagesApiRoute({