Merge pull request #3056 from arimu1/fix/3052-fetch-readablestream-body

fix(net): buffer closed ReadableStream bodies for fetch POST

src/browser/tests/net/fetch.html

@@ -348,6 +348,93 @@
   }
 </script>
 
+<script id=fetch_readablestream_body type=module>
+  {
+    const state = await testing.async();
+    const payload = "name=Streamed&count=9";
+    const enc = new TextEncoder();
+    const stream = new ReadableStream({
+      start(controller) {
+        controller.enqueue(enc.encode("name=Streamed&"));
+        controller.enqueue(enc.encode("count=9"));
+        controller.close();
+      },
+    });
+
+    const response = await fetch("http://127.0.0.1:9582/echo_body", {
+      method: "POST",
+      body: stream,
+      duplex: "half",
+      headers: { "Content-Type": "application/x-www-form-urlencoded" },
+    });
+    const text = await response.text();
+    state.resolve();
+
+    await state.done(() => {
+      testing.expectEqual(200, response.status);
+      testing.expectEqual(payload, text);
+    });
+  }
+</script>
+
+<script id=fetch_readablestream_body_reject type=module>
+  {
+    const state = await testing.async();
+    const url = "http://127.0.0.1:9582/echo_body";
+    const opts = { method: "POST", duplex: "half" };
+
+    let openRejected = false;
+    try {
+      const open = new ReadableStream({
+        start(c) { c.enqueue(new TextEncoder().encode("x")); },
+      });
+      await fetch(url, { ...opts, body: open });
+    } catch (e) {
+      openRejected = e instanceof TypeError;
+    }
+
+    let lockedRejected = false;
+    try {
+      const locked = new ReadableStream({
+        start(c) {
+          c.enqueue(new TextEncoder().encode("x"));
+          c.close();
+        },
+      });
+      locked.getReader();
+      await fetch(url, { ...opts, body: locked });
+    } catch (e) {
+      lockedRejected = e instanceof TypeError;
+    }
+
+    // A stream is single-use: draining it for one body must not leave it
+    // looking like a valid (empty) body for the next request.
+    const once = new ReadableStream({
+      start(c) {
+        c.enqueue(new TextEncoder().encode("x"));
+        c.close();
+      },
+    });
+    const first = await fetch(url, { ...opts, body: once });
+    const firstText = await first.text();
+
+    let reuseRejected = false;
+    try {
+      await fetch(url, { ...opts, body: once });
+    } catch (e) {
+      reuseRejected = e instanceof TypeError;
+    }
+
+    state.resolve();
+    await state.done(() => {
+      testing.expectEqual(true, openRejected);
+      testing.expectEqual(true, lockedRejected);
+      testing.expectEqual("x", firstText);
+      testing.expectEqual(true, reuseRejected);
+    });
+  }
+</script>
+
 <script id=fetch_blob_url_in_worker type=module>
   {
     const state = await testing.async();

src/browser/webapi/net/body_init.zig

@@ -104,14 +104,13 @@ pub const BodyInit = union(enum) {
                     .content_type = null,
                 };
             },
-            .stream => {
-                // A ReadableStream body cannot be serialized synchronously.
-                // Callers that support streaming bodies (Response) special-case
-                // the `.stream` arm before calling extract; the Request/XHR
-                // paths that reach here have no place to store a stream, so
-                // they send an empty body. Like other non-string sources, a
-                // stream carries no default Content-Type.
-                return .{ .bytes = "", .content_type = null };
+            .stream => |stream| {
+                // Response special-cases `.stream` before extract. Request/XHR
+                // paths buffer a closed stream synchronously; a stream that
+                // can't be drained here - still open, or already used as a
+                // body - rejects rather than send Content-Length: 0.
+                const bytes = try stream.collectBodyBytes(arena);
+                return .{ .bytes = bytes, .content_type = null };
             },
         }
     }

src/browser/webapi/streams/ReadableStream.zig

@@ -53,6 +53,7 @@ _pull_fn: ?js.Function.Global = null,
 _pulling: bool = false,
 _pull_again: bool = false,
 _cancel: ?Cancel = null,
+_collected: bool = false,
 
 const UnderlyingSource = struct {
     start: ?js.Function = null,
@@ -131,6 +132,48 @@ pub fn getLocked(self: *const ReadableStream) bool {
     return self._reader != null;
 }
 
+/// Drain a closed stream's internal queue into bytes for outbound HTTP
+/// request bodies. A stream that is locked, already used as a body, still
+/// readable, or errored - or that holds a chunk which isn't string or binary
+/// data - is `error.TypeError`. The queue is only consumed once every chunk
+/// has converted, so a rejected body leaves the stream as it was.
+pub fn collectBodyBytes(self: *ReadableStream, arena: std.mem.Allocator) ![]const u8 {
+    if (self._collected or self.getLocked()) {
+        return error.TypeError;
+    }
+    switch (self._state) {
+        .errored, .readable => return error.TypeError,
+        .closed => {},
+    }
+
+    const local = self._execution.js.local.?;
+
+    var buf = std.Io.Writer.Allocating.init(arena);
+    const queue = &self._controller._queue;
+    for (queue.items) |chunk| {
+        const bytes: []const u8 = switch (chunk) {
+            .string => |s| s,
+            .uint8array => |arr| arr.values,
+            .js_value => |global| blk: {
+                const value = local.toLocal(global);
+                // toStringSmart falls back to toString(), which would turn a
+                // stray number or object into "42" / "[object Object]" bytes.
+                if (value.isString() == null and !value.isTypedArray() and
+                    !value.isArrayBufferView() and !value.isArrayBuffer())
+                {
+                    return error.TypeError;
+                }
+                break :blk try value.toStringSmart();
+            },
+        };
+        try buf.writer.writeAll(bytes);
+    }
+
+    self._collected = true;
+    queue.clearRetainingCapacity();
+    return buf.written();
+}
+
 pub fn callPullIfNeeded(self: *ReadableStream) !void {
     if (!self.shouldCallPull()) {
         return;

src/testing.zig

@@ -988,6 +988,21 @@ fn testHTTPHandler(req: *std.http.Server.Request) !void {
         });
     }
 
+    if (std.mem.eql(u8, path, "/echo_body")) {
+        // Echo the request body back verbatim, so tests can assert on the bytes
+        // a request actually sent rather than just on its status.
+        var body_buf: [4096]u8 = undefined;
+        const body = if (req.head.method.requestHasBody())
+            try req.readerExpectNone(&body_buf).allocRemaining(arena_allocator, .limited(body_buf.len))
+        else
+            "";
+        return req.respond(body, .{
+            .extra_headers = &.{
+                .{ .name = "Content-Type", .value = "text/plain; charset=utf-8" },
+            },
+        });
+    }
+
     if (std.mem.eql(u8, path, "/redirect_to_echo")) {
         // 302 to /echo_method. Used by the Page.reload-after-redirect test to
         // confirm a POST→302→GET chain doesn't replay POST on reload.