perf(ext/fetch): use content-length in InnerBody.consume (#15925)

This fast path prevents repeated allocations when receiving a fetch body with a known size.

Co-authored-by: Luca Casonato <hello@lcas.dev>

cli/tests/unit/fetch_test.ts

@@ -1639,3 +1639,137 @@ Deno.test(async function staticResponseJson() {
   const res = await resp.json();
   assertEquals(res, data);
 });
+
+function invalidServer(addr: string, body: Uint8Array): Deno.Listener {
+  const [hostname, port] = addr.split(":");
+  const listener = Deno.listen({
+    hostname,
+    port: Number(port),
+  }) as Deno.Listener;
+
+  (async () => {
+    for await (const conn of listener) {
+      const p1 = conn.read(new Uint8Array(2 ** 14));
+      const p2 = conn.write(body);
+
+      await Promise.all([p1, p2]);
+      conn.close();
+    }
+  })();
+
+  return listener;
+}
+
+Deno.test(
+  { permissions: { net: true } },
+  async function fetchWithInvalidContentLengthAndTransferEncoding(): Promise<
+    void
+  > {
+    const addr = "127.0.0.1:4516";
+    const data = "a".repeat(10 << 10);
+
+    const body = new TextEncoder().encode(
+      `HTTP/1.1 200 OK\r\nContent-Length: ${
+        Math.round(data.length * 2)
+      }\r\nTransfer-Encoding: chunked\r\n\r\n${
+        data.length.toString(16)
+      }\r\n${data}\r\n0\r\n\r\n`,
+    );
+
+    // if transfer-encoding is sent, content-length is ignored
+    // even if it has an invalid value (content-length > totalLength)
+    const listener = invalidServer(addr, body);
+    const response = await fetch(`http://${addr}/`);
+
+    const res = await response.arrayBuffer();
+    const buf = new TextEncoder().encode(data);
+    assertEquals(res.byteLength, buf.byteLength);
+    assertEquals(new Uint8Array(res), buf);
+
+    listener.close();
+  },
+);
+
+Deno.test(
+  { permissions: { net: true } },
+  async function fetchWithInvalidContentLength(): Promise<
+    void
+  > {
+    const addr = "127.0.0.1:4517";
+    const data = "a".repeat(10 << 10);
+
+    const body = new TextEncoder().encode(
+      `HTTP/1.1 200 OK\r\nContent-Length: ${
+        Math.round(data.length / 2)
+      }\r\nContent-Length: ${data.length}\r\n\r\n${data}`,
+    );
+
+    // It should fail if multiple content-length headers with different values are sent
+    const listener = invalidServer(addr, body);
+    await assertRejects(
+      async () => {
+        await fetch(`http://${addr}/`);
+      },
+      TypeError,
+      "invalid content-length parsed",
+    );
+
+    listener.close();
+  },
+);
+
+Deno.test(
+  { permissions: { net: true } },
+  async function fetchWithInvalidContentLength(): Promise<
+    void
+  > {
+    const addr = "127.0.0.1:4518";
+    const data = "a".repeat(10 << 10);
+
+    const contentLength = data.length / 2;
+    const body = new TextEncoder().encode(
+      `HTTP/1.1 200 OK\r\nContent-Length: ${contentLength}\r\n\r\n${data}`,
+    );
+
+    const listener = invalidServer(addr, body);
+    const response = await fetch(`http://${addr}/`);
+
+    // If content-length < totalLength, a maximum of content-length bytes
+    // should be returned.
+    const res = await response.arrayBuffer();
+    const buf = new TextEncoder().encode(data);
+    assertEquals(res.byteLength, contentLength);
+    assertEquals(new Uint8Array(res), buf.subarray(contentLength));
+
+    listener.close();
+  },
+);
+
+Deno.test(
+  { permissions: { net: true } },
+  async function fetchWithInvalidContentLength(): Promise<
+    void
+  > {
+    const addr = "127.0.0.1:4519";
+    const data = "a".repeat(10 << 10);
+
+    const contentLength = data.length * 2;
+    const body = new TextEncoder().encode(
+      `HTTP/1.1 200 OK\r\nContent-Length: ${contentLength}\r\n\r\n${data}`,
+    );
+
+    const listener = invalidServer(addr, body);
+    const response = await fetch(`http://${addr}/`);
+    // If content-length > totalLength, a maximum of content-length bytes
+    // should be returned.
+    await assertRejects(
+      async () => {
+        await response.arrayBuffer();
+      },
+      Error,
+      "end of file before message length reached",
+    );
+
+    listener.close();
+  },
+);

ext/fetch/22_body.js

@@ -64,10 +64,12 @@
   }
 
   class InnerBody {
+    #knownExactLength = null;
+
     /**
      * @param {ReadableStream<Uint8Array> | { body: Uint8Array | string, consumed: boolean }} stream
      */
-    constructor(stream) {
+    constructor(stream, knownExactLength) {
       /** @type {ReadableStream<Uint8Array> | { body: Uint8Array | string, consumed: boolean }} */
       this.streamOrStatic = stream ??
         { body: new Uint8Array(), consumed: false };
@@ -75,6 +77,8 @@
       this.source = null;
       /** @type {null | number} */
       this.length = null;
+
+      this.#knownExactLength = knownExactLength;
     }
 
     get stream() {
@@ -147,14 +151,31 @@
         const reader = this.stream.getReader();
         /** @type {Uint8Array[]} */
         const chunks = [];
+
+        let finalBuffer = this.#knownExactLength
+          ? new Uint8Array(this.#knownExactLength)
+          : null;
+
         let totalLength = 0;
         while (true) {
           const { value: chunk, done } = await reader.read();
           if (done) break;
+
+          if (finalBuffer) {
+            // fast path, content-length is present
+            TypedArrayPrototypeSet(finalBuffer, chunk, totalLength);
+          } else {
+            // slow path, content-length is not present
             ArrayPrototypePush(chunks, chunk);
+          }
           totalLength += chunk.byteLength;
         }
-        const finalBuffer = new Uint8Array(totalLength);
+
+        if (finalBuffer) {
+          return finalBuffer;
+        }
+
+        finalBuffer = new Uint8Array(totalLength);
         let i = 0;
         for (const chunk of chunks) {
           TypedArrayPrototypeSet(finalBuffer, chunk, i);
@@ -199,7 +220,7 @@
     clone() {
       const [out1, out2] = this.stream.tee();
       this.streamOrStatic = out1;
-      const second = new InnerBody(out2);
+      const second = new InnerBody(out2, this.#knownExactLength);
       second.source = core.deserialize(core.serialize(this.source));
       second.length = this.length;
       return second;

ext/fetch/26_fetch.js

@@ -335,6 +335,7 @@
       } else {
         response.body = new InnerBody(
           createResponseBodyStream(resp.responseRid, terminator),
+          resp.contentLength,
         );
       }
     }

ext/fetch/lib.rs

@@ -361,6 +361,7 @@ pub struct FetchResponse {
   headers: Vec<(ByteString, ByteString)>,
   url: String,
   response_rid: ResourceId,
+  content_length: Option<u64>,
 }
 
 #[op]
@@ -391,6 +392,8 @@ pub async fn op_fetch_send(
     res_headers.push((key.as_str().into(), val.as_bytes().into()));
   }
 
+  let content_length = res.content_length();
+
   let stream: BytesStream = Box::pin(res.bytes_stream().map(|r| {
     r.map_err(|err| std::io::Error::new(std::io::ErrorKind::Other, err))
   }));
@@ -409,6 +412,7 @@ pub async fn op_fetch_send(
     headers: res_headers,
     url,
     response_rid: rid,
+    content_length,
   })
 }