fetch: deliver the chunks ahead of a malformed chunk-size before the error

src/http/lib.rs

@@ -1596,6 +1596,7 @@ impl<'a> HTTPClient<'a> {
     /// Common tail of `fail` / `fail_from_h2` / `complete_connecting_process`:
     /// build the result, reset request state, and dispatch the callback.
     fn dispatch_result_and_reset(&mut self, clear_proxy_tunneling: bool) {
+        self.report_body_decoded_before_failure();
         let callback = self.result_callback;
         let result = self.to_result();
         self.state.reset();
@@ -1617,6 +1618,36 @@ impl<'a> HTTPClient<'a> {
         }
         callback.run(self.parent_async_http(), result);
     }
+    /// Body bytes decoded ahead of the failure reach a streaming consumer first, in a progress callback.
+    fn report_body_decoded_before_failure(&mut self) {
+        let is_streaming = self.signals.get(signals::Field::ResponseBodyStreaming)
+            || self.signals.body_receive_mode.is_some();
+        if !is_streaming
+            || self.state.decoded_body.list.is_empty()
+            || self.state.flags.is_redirect_pending
+        {
+            return;
+        }
+        // An abort or a timeout ends the body where the consumer stands.
+        let fail = match self.state.fail.take() {
+            None => return,
+            Some(
+                fail @ (crate::Error::Aborted
+                | crate::Error::AbortedBeforeConnecting
+                | crate::Error::Timeout),
+            ) => {
+                self.state.fail = Some(fail);
+                return;
+            }
+            Some(fail) => fail,
+        };
+        let mut result = self.to_result();
+        self.state.fail = Some(fail);
+        result.has_more = true;
+        let decoded_body = core::mem::take(&mut self.state.decoded_body);
+        result.body = decoded_body.list.as_slice();
+        self.result_callback.run(self.parent_async_http(), result);
+    }
     #[inline]
     fn progress_node_mut(&mut self) -> Option<&mut bun_core::Progress::Node> {
         // SAFETY: progress_node is owned by the caller (e.g. `bun install`'s
@@ -4621,11 +4652,14 @@ impl<'a> HTTPClient<'a> {
         );
 
         match pret {
-            // Invalid HTTP response body
-            -1 => return Err(crate::Error::InvalidHTTPResponse),
-            // Needs more data
-            -2 => {
+            // Invalid HTTP response body. Node.js gives a compressed body nothing from this read.
+            -1 if self.state.encoding.is_compressed() => {
+                return Err(crate::Error::InvalidHTTPResponse);
+            }
+            // -1: invalid, -2: needs more data. The chunks decoded so far are body either way.
+            -1 | -2 => {
                 self.report_progress(buffer_len);
+                let mut processed = false;
                 // streaming chunks
                 if self.signals.get(signals::Field::ResponseBodyStreaming)
                     || self.signals.body_receive_mode.is_some()
@@ -4635,10 +4669,13 @@ impl<'a> HTTPClient<'a> {
                     // Move the
                     // bytes out so no `&` into self.state aliases the `&mut self.state` call.
                     let buffer_snap = core::mem::take(&mut self.state.get_body_buffer().list);
-                    return self.state.process_body_buffer(buffer_snap, false);
+                    processed = self.state.process_body_buffer(buffer_snap, false)?;
+                }
+                if pret == -1 {
+                    return Err(crate::Error::InvalidHTTPResponse);
                 }
 
-                return Ok(false);
+                return Ok(processed);
             }
             // Done
             _ => {
@@ -4695,13 +4732,14 @@ impl<'a> HTTPClient<'a> {
             self.state.total_body_received
         );
         match pret {
-            // Invalid HTTP response body
-            -1 => Err(crate::Error::InvalidHTTPResponse),
-            // Needs more data
-            -2 => {
+            // Invalid HTTP response body. Node.js gives a compressed body nothing from this read.
+            -1 if self.state.encoding.is_compressed() => Err(crate::Error::InvalidHTTPResponse),
+            // -1: invalid, -2: needs more data. The chunks decoded so far are body either way.
+            -1 | -2 => {
                 self.report_progress(buffer.len());
                 self.state.get_body_buffer().append_slice_exact(buffer)?;
 
+                let mut processed = false;
                 // streaming chunks
                 if self.signals.get(signals::Field::ResponseBodyStreaming)
                     || self.signals.body_receive_mode.is_some()
@@ -4713,10 +4751,13 @@ impl<'a> HTTPClient<'a> {
                     // the bytes out so no `&` into self.state aliases the `&mut self.state`
                     // taken by process_body_buffer (which mutates compressed_body/decoded_body).
                     let buffer_snap = core::mem::take(&mut self.state.get_body_buffer().list);
-                    return self.state.process_body_buffer(buffer_snap, false);
+                    processed = self.state.process_body_buffer(buffer_snap, false)?;
+                }
+                if pret == -1 {
+                    return Err(crate::Error::InvalidHTTPResponse);
                 }
 
-                Ok(false)
+                Ok(processed)
             }
             // Done
             _ => {

src/runtime/webcore/fetch/FetchTasklet.rs

@@ -881,14 +881,44 @@ impl FetchTasklet {
         Ok(())
     }
 
+    /// Takes a failure that arrived behind body bytes this thread has not seen out of `result`.
+    fn hold_failure_behind_unseen_body(&mut self) -> Option<http::Error> {
+        let fail = self.result.fail?;
+        if self.scheduled_response_buffer.list.is_empty() {
+            return None;
+        }
+        // Without a head the promise rejects and nothing can read the bytes.
+        if !self.is_waiting_body && self.metadata.is_none() {
+            return None;
+        }
+        // An abort ends the body where the consumer stands.
+        if self.is_waiting_abort
+            || self.abort_reason.has()
+            || self.result.abort_reason().is_some()
+            || self.signal_store.aborted.load(Ordering::Relaxed)
+        {
+            return None;
+        }
+        self.result.fail = None;
+        self.result.has_more = true;
+        Some(fail)
+    }
+
     pub(crate) fn on_progress_update(&mut self) -> JsResult<()> {
         jsc::mark_binding!();
         bun_output::scoped_log!(FetchTasklet, "onProgressUpdate");
         self.mutex.lock();
-        self.has_schedule_callback.store(false, Ordering::Relaxed);
-        let is_done = !self.result.has_more;
+        // Set by `callback` only: this run was posted by the HTTP thread.
+        let posted_by_http_thread = self.has_schedule_callback.swap(false, Ordering::Relaxed);
 
         let vm = self.global_this.bun_vm();
+        let failure_behind_body = if posted_by_http_thread && vm.script_allowed() {
+            self.hold_failure_behind_unseen_body()
+        } else {
+            None
+        };
+        let is_done = !self.result.has_more;
+
         // teardown forbade script: we cannot touch JS
         if !vm.script_allowed() {
             // The certificate will never be checked; release the parked
@@ -910,6 +940,15 @@ impl FetchTasklet {
         let global_this = self.global_this;
         // explicit cleanup at each return (a closure keeps borrowck happy)
         let cleanup = |this: &mut FetchTasklet| {
+            if let Some(fail) = failure_behind_body {
+                // The next task handles the failure as the terminal update, and owns this thread's ref.
+                this.result.fail = Some(fail);
+                this.result.has_more = false;
+                // SAFETY: `vm.event_loop()` is the live JS-thread loop.
+                unsafe {
+                    (*vm.event_loop()).enqueue_task(Task::init(std::ptr::from_mut(this)));
+                }
+            }
             this.mutex.unlock();
             // if we are not done we wait until the next call
             if is_done {
@@ -929,8 +968,9 @@ impl FetchTasklet {
             // start streaming
             if let Err(err) = self.start_request_stream() {
                 // The VM is being stopped: leave like the `!script_allowed()` gate above does.
+                // A failure held back for a second run is the terminal update; no run follows.
                 self.mutex.unlock();
-                if is_done {
+                if is_done || failure_behind_body.is_some() {
                     // SAFETY: `self` is the live heap tasklet; we hold a ref.
                     FetchTasklet::deref(std::ptr::from_mut(self));
                 }

test/js/web/fetch/fetch-chunked-size.test.ts

@@ -37,6 +37,104 @@ describe("fetch: chunked chunk-size token validation", () => {
     });
   });
 
+  // The chunks ahead of the malformed one are body. A reader that reads at once gets them,
+  // then the read rejects, whether or not they share a socket read with the malformed line.
+  // node v26.3.0: chunks ["x"], then `TypeError: terminated` (HPE_INVALID_CHUNK_SIZE).
+  describe("delivers the chunks ahead of a malformed chunk-size to a streaming reader", () => {
+    const head = "HTTP/1.1 200 OK\r\nTransfer-Encoding: chunked\r\n";
+    // "C" is a hex digit, so this line is chunk-size 0xC followed by junk.
+    const malformed = "Connection: close\r\n\r\n";
+
+    // Writes `first` when the request arrives and keeps the socket open, so the failure can
+    // only come from the parser. `send()` writes one more piece.
+    async function serveParts(first: string | Buffer) {
+      const { promise, resolve } = Promise.withResolvers<net.AddressInfo>();
+      const accepted = Promise.withResolvers<net.Socket>();
+      const server = net
+        .createServer(socket => {
+          socket.on("error", () => {});
+          socket.once("data", () => {
+            socket.write(first);
+            accepted.resolve(socket);
+          });
+        })
+        .listen(0, "127.0.0.1", () => resolve(server.address() as net.AddressInfo));
+      const address = await promise;
+      return {
+        server,
+        url: `http://127.0.0.1:${address.port}/`,
+        send: async (part: string | Buffer) => void (await accepted.promise).write(part),
+      };
+    }
+
+    async function readUntilError(reader: ReadableStreamDefaultReader<Uint8Array>) {
+      const chunks: Buffer[] = [];
+      try {
+        while (true) {
+          const { done, value } = await reader.read();
+          if (done) return { body: Buffer.concat(chunks).toString(), code: "(stream ended)" };
+          chunks.push(Buffer.from(value));
+        }
+      } catch (e: any) {
+        return { body: Buffer.concat(chunks).toString(), code: e?.code };
+      }
+    }
+
+    it("in the same read as the response head", async () => {
+      const { server, url } = await serveParts(`${head}\r\n1\r\nx\r\n${malformed}`);
+      await using _s = server;
+      const res = await fetch(url);
+      expect(await readUntilError(res.body!.getReader())).toEqual({ body: "x", code: "InvalidHTTPResponse" });
+    });
+
+    it("in a read after the response head", async () => {
+      const { server, url, send } = await serveParts(`${head}\r\n`);
+      await using _s = server;
+      const res = await fetch(url);
+      // A read is pending before the server writes the rest.
+      const result = readUntilError(res.body!.getReader());
+      await send(`1\r\nx\r\n${malformed}`);
+      expect(await result).toEqual({ body: "x", code: "InvalidHTTPResponse" });
+    });
+
+    // node v26.3.0 delivers nothing here: the error tears its gunzip down before it emits.
+    it("but not of a compressed body", async () => {
+      const gz = Bun.gzipSync("hello hello hello hello");
+      const wire = Buffer.concat([
+        Buffer.from(`${head}Content-Encoding: gzip\r\n\r\n${gz.length.toString(16)}\r\n`),
+        gz,
+        Buffer.from(`\r\n${malformed}`),
+      ]);
+      const { server, url } = await serveParts(wire);
+      await using _s = server;
+      const res = await fetch(url);
+      expect(await readUntilError(res.body!.getReader())).toEqual({ body: "", code: "InvalidHTTPResponse" });
+    });
+
+    it("with a chunk larger than one socket read", async () => {
+      const payload = Buffer.alloc(256 * 1024, "abcdefghijklmnopqrstuvwxyz").toString();
+      const { server, url } = await serveParts(
+        `${head}\r\n${payload.length.toString(16)}\r\n${payload}\r\n${malformed}`,
+      );
+      await using _s = server;
+      const res = await fetch(url);
+      const { body, code } = await readUntilError(res.body!.getReader());
+      expect(code).toBe("InvalidHTTPResponse");
+      expect(body.length).toBe(payload.length);
+      expect(body).toBe(payload);
+    });
+
+    it("text() still rejects", async () => {
+      const { server, url } = await serveParts(`${head}\r\n1\r\nx\r\n${malformed}`);
+      await using _s = server;
+      const result = await fetch(url)
+        .then(res => res.text())
+        .then(body => ({ resolved: body }))
+        .catch(e => e);
+      expect(result?.code).toBe("InvalidHTTPResponse");
+    });
+  });
+
   describe("accepts well-formed chunk-size", () => {
     it.each([
       ["5", "hello"],