Prevent double drain

Use existing toStringSmart to convert value to a string

src/browser/tests/net/fetch.html

@@ -361,7 +361,7 @@
       },
     });
 
-    const response = await fetch("http://127.0.0.1:9582/xhr/echo_post_body", {
+    const response = await fetch("http://127.0.0.1:9582/echo_body", {
       method: "POST",
       body: stream,
       duplex: "half",
@@ -380,7 +380,7 @@
 <script id=fetch_readablestream_body_reject type=module>
   {
     const state = await testing.async();
-    const url = "http://127.0.0.1:9582/xhr/echo_post_body";
+    const url = "http://127.0.0.1:9582/echo_body";
     const opts = { method: "POST", duplex: "half" };
 
     let openRejected = false;
@@ -407,10 +407,30 @@
       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>

src/browser/webapi/net/body_init.zig

@@ -106,8 +106,9 @@ pub const BodyInit = union(enum) {
             },
             .stream => |stream| {
                 // Response special-cases `.stream` before extract. Request/XHR
-                // paths buffer a closed stream synchronously; open streams and
-                // async-only bodies reject rather than send Content-Length: 0.
+                // 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,
@@ -132,39 +133,44 @@ pub fn getLocked(self: *const ReadableStream) bool {
 }
 
 /// Drain a closed stream's internal queue into bytes for outbound HTTP
-/// request bodies. Open, errored, or locked streams, and chunks that cannot
-/// be converted synchronously, return `error.TypeError`.
+/// 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.getLocked()) return error.TypeError;
+    if (self._collected or self.getLocked()) {
+        return error.TypeError;
+    }
     switch (self._state) {
         .errored, .readable => return error.TypeError,
         .closed => {},
     }
 
-    const exec = self._execution;
-    const local = exec.js.local orelse return error.TypeError;
+    const local = self._execution.js.local.?;
 
     var buf = std.Io.Writer.Allocating.init(arena);
-    const controller = self._controller;
-    while (controller.dequeue()) |chunk| {
+    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);
-                if (value.isTypedArray() or value.isArrayBufferView() or value.isArrayBuffer()) {
-                    const typed = try local.jsValueToZig([]u8, value);
-                    break :blk try arena.dupe(u8, typed);
-                } else if (value.isString()) |str| {
-                    const slice = try str.toSlice();
-                    break :blk try arena.dupe(u8, slice);
-                } else {
+                // 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();
 }
 

src/testing.zig

@@ -970,10 +970,12 @@ fn testHTTPHandler(req: *std.http.Server.Request) !void {
         });
     }
 
-    if (std.mem.eql(u8, path, "/xhr/echo_post_body")) {
+    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())
-            req.readerExpectNone(&body_buf).allocRemaining(arena_allocator, .limited(body_buf.len)) catch ""
+            try req.readerExpectNone(&body_buf).allocRemaining(arena_allocator, .limited(body_buf.len))
         else
             "";
         return req.respond(body, .{