fix(agents): stream forwarded request bodies into sub-agents

.changeset/facet-forward-stream-body.md

@@ -0,0 +1,22 @@
+---
+"agents": patch
+---
+
+Stream forwarded request bodies into sub-agents instead of buffering them in the parent Durable Object.
+
+`Agent._cf_forwardToFacet` and `routeSubAgentRequest` both did `forwardInit.body = await req.arrayBuffer()` before dispatching to a child facet, materialising the entire request body in the parent's isolate. Two consequences:
+
+- The read sat **in front of** application-level validation. `Agent.fetch` returns before `onRequest` whenever the path matches `/sub/{class}/{name}`, so an app that carefully bounded request bodies in `onRequest` still had an unbounded read ahead of it — and no way to bound it itself.
+- The cost was **per hop**. A nested `/sub/.../sub/...` address re-materialised the same bytes at every level.
+
+Both call sites now pass `req.body` through as a stream. Measured on `wrangler dev --local` with a handler that never reads the body, peak RSS across the `workerd` processes for a single POST:
+
+| Request body | facet route, before | facet route, after | canonical route (control) |
+| ------------ | ------------------- | ------------------ | ------------------------- |
+| 16 MB        | +75 MB              | +4 MB              | +2 MB                     |
+| 64 MB        | +268 MB             | +4 MB              | +2 MB                     |
+| 128 MB       | +546 MB             | +4 MB              | +2 MB                     |
+
+This restores the behaviour from before #1443, which switched to an explicit `RequestInit` in order to set a header on WebSocket upgrades and re-attached the body with `arrayBuffer()` as a side effect. The `Upgrade` header handling from that fix is unchanged.
+
+One behavioural note: backpressure now reaches the client. A child that returns without reading the body will cause the remainder of the upload to be cancelled, where previously the parent drained it in full.

packages/agents/src/index.ts

@@ -7736,8 +7736,12 @@ export class Agent<
     if (req.headers.get("Upgrade")?.toLowerCase() === "websocket") {
       forwardedHeaders.set(SUB_AGENT_OUTER_URL_HEADER, req.url);
     }
+    // Hand the body through as a stream. Reading it here (e.g.
+    // `await req.arrayBuffer()`) materialises the entire body in the
+    // parent DO's isolate, ahead of any application-level intake limit,
+    // and re-materialises it once per `/sub/` hop — see #2015.
     if (req.body && req.method !== "GET" && req.method !== "HEAD") {
-      forwardedInit.body = await req.arrayBuffer();
+      forwardedInit.body = req.body;
     }
     const forwarded = new Request(rewritten, forwardedInit);
     return fetcher.fetch(forwarded);

packages/agents/src/sub-routing.ts

@@ -424,8 +424,12 @@ export async function routeSubAgentRequest(
     method: req.method,
     headers: new Headers(req.headers)
   };
+  // Stream the body through rather than buffering it — see #2015. This
+  // helper runs in the caller's isolate, but the same request then hits
+  // `_cf_forwardToFacet` on the parent, so buffering here would put a
+  // second unbounded copy in front of the child.
   if (req.body && req.method !== "GET" && req.method !== "HEAD") {
-    forwardInit.body = await req.arrayBuffer();
+    forwardInit.body = req.body;
   }
   const forwardReq = new Request(forwardUrl, forwardInit);
 

packages/agents/src/tests/agents/index.ts

@@ -71,5 +71,7 @@ export {
   Sub_,
   ReservedClassParent,
   TestUnboundParentAgent,
-  TestMinifiedNameParentAgent
+  TestMinifiedNameParentAgent,
+  BodyProbeSubAgent,
+  BodyProbeRootAgent
 } from "./sub-agent";

packages/agents/src/tests/agents/sub-agent.ts

@@ -2223,3 +2223,100 @@ class _a extends Agent {
   }
 }
 export { _a as TestMinifiedNameParentAgent };
+
+// ── Request-body forwarding probes (issue #2015) ─────────────────────
+//
+// `_cf_forwardToFacet` and `routeSubAgentRequest` used to materialise
+// the whole forwarded body via `await req.arrayBuffer()` before
+// dispatching. These fixtures let a test observe *when* the child sees
+// the request relative to the client finishing its upload, which is
+// what distinguishes streaming from buffering.
+//
+// The same handler backs a facet child (`BodyProbeSubAgent`) and a
+// root Agent (`BodyProbeRootAgent`). The root is the control: it
+// proves the *test harness* can stream a request body, so a hang on
+// the facet path can be attributed to the forwarder rather than to
+// vitest-pool-workers.
+
+function toHex(buf: ArrayBuffer): string {
+  return Array.from(new Uint8Array(buf))
+    .map((b) => b.toString(16).padStart(2, "0"))
+    .join("");
+}
+
+/**
+ * Probe endpoints:
+ *   - `/probe/ignore`      — reply immediately, never touch the body.
+ *   - `/probe/first-chunk` — read exactly one chunk, then reply.
+ *   - `/probe/drain`       — read to completion, reply with size + digest.
+ */
+async function handleBodyProbe(
+  agentName: string,
+  request: Request
+): Promise<Response> {
+  const { pathname } = new URL(request.url);
+
+  // Matched by suffix, not equality: a facet child sees the tail after
+  // `/sub/{class}/{name}` (the forwarder rewrites the pathname), while a
+  // root Agent sees the full `/agents/{class}/{name}/...` path. The same
+  // handler has to serve both.
+  if (pathname.endsWith("/probe/ignore")) {
+    return Response.json({ agentName, probe: "ignore" });
+  }
+
+  if (pathname.endsWith("/probe/first-chunk")) {
+    if (!request.body) {
+      return Response.json({ agentName, chunk: null, probe: "first-chunk" });
+    }
+    const reader = request.body.getReader();
+    try {
+      const { done, value } = await reader.read();
+      return Response.json({
+        agentName,
+        chunk: value ? new TextDecoder().decode(value) : null,
+        done,
+        probe: "first-chunk"
+      });
+    } finally {
+      // Let go without draining — the point of this probe is that the
+      // child can act on a prefix of a body the client hasn't finished
+      // sending.
+      reader.cancel().catch(() => {});
+    }
+  }
+
+  if (pathname.endsWith("/probe/drain")) {
+    const body = await request.arrayBuffer();
+    return Response.json({
+      agentName,
+      bytes: body.byteLength,
+      contentLength: request.headers.get("content-length"),
+      probe: "drain",
+      sha256: toHex(await crypto.subtle.digest("SHA-256", body))
+    });
+  }
+
+  return Response.json(
+    { agentName, path: pathname, probe: "unknown" },
+    {
+      status: 404
+    }
+  );
+}
+
+/** Facet-only child. Reached via `/sub/body-probe-sub-agent/{name}`. */
+export class BodyProbeSubAgent extends Agent {
+  override async onRequest(request: Request): Promise<Response> {
+    return handleBodyProbe(this.name, request);
+  }
+}
+
+/**
+ * Root Agent running the identical handler — the canonical (non-facet)
+ * control path from the issue's measurement table.
+ */
+export class BodyProbeRootAgent extends Agent {
+  override async onRequest(request: Request): Promise<Response> {
+    return handleBodyProbe(this.name, request);
+  }
+}