Throw instead of aborting when stream text() consumers accumulate more than 2^31-1 bytes

src/jsc/bindings/webcore/streams/BunStreamConsumers.cpp

@@ -283,6 +283,23 @@ static size_t writeUTF8(const WTF::String& string, std::span<uint8_t> destinatio
     return Bun__encoding__writeUTF16(string.span16().data(), string.span16().size(), destination.data(), destination.size(), utf8);
 }
 
+bool appendUTF8WithinStringLimit(const WTF::String& string, WTF::Vector<uint8_t>& bytes)
+{
+    size_t byteLength = utf8ByteLengthWithReplacement(string);
+    if (!byteLength)
+        return true;
+    size_t oldSize = bytes.size();
+    // UTF-8 expansion can exceed any reserve taken from the code-unit estimate.
+    if (exceedsStringLimit(oldSize + byteLength) || !bytes.tryGrow(oldSize + byteLength)) [[unlikely]]
+        return false;
+    size_t written = writeUTF8(string, bytes.mutableSpan().subspan(oldSize));
+    // The sizer and writer must agree; never expose ungrown (uninitialized) bytes.
+    ASSERT(written == byteLength);
+    if (written < byteLength) [[unlikely]]
+        bytes.shrink(oldSize + written);
+    return true;
+}
+
 // `obj[name](...args)` with `this` = obj.
 static JSValue invokeMethod(JSC::VM& vm, JSGlobalObject* globalObject, JSObject* object, const Identifier& name, const MarkedArgumentBuffer& args)
 {
@@ -325,25 +342,26 @@ static bool appendChunkBytes(JSC::VM& vm, JSGlobalObject* globalObject, JSValue
     if (chunk.isString()) {
         WTF::String string = asString(chunk)->value(globalObject);
         RETURN_IF_EXCEPTION(scope, false);
-        if (size_t byteLength = utf8ByteLengthWithReplacement(string)) {
-            size_t oldSize = bytes.size();
-            bytes.grow(oldSize + byteLength);
-            size_t written = writeUTF8(string, bytes.mutableSpan().subspan(oldSize));
-            // The sizer and writer must agree; never expose ungrown (uninitialized) bytes.
-            ASSERT(written == byteLength);
-            if (written < byteLength) [[unlikely]]
-                bytes.shrink(oldSize + written);
+        if (!appendUTF8WithinStringLimit(string, bytes)) [[unlikely]] {
+            throwOutOfMemoryError(globalObject, scope);
+            return false;
         }
         return true;
     }
     if (auto* view = dynamicDowncast<JSC::JSArrayBufferView>(chunk)) {
-        if (!view->isDetached())
-            bytes.append(view->span());
+        if (!view->isDetached() && !bytes.tryAppend(view->span())) [[unlikely]] {
+            throwOutOfMemoryError(globalObject, scope);
+            return false;
+        }
         return true;
     }
     if (auto* jsBuffer = dynamicDowncast<JSC::JSArrayBuffer>(chunk)) {
-        if (auto* impl = jsBuffer->impl(); impl && !impl->isDetached())
-            bytes.append(impl->span());
+        if (auto* impl = jsBuffer->impl(); impl && !impl->isDetached()) {
+            if (!bytes.tryAppend(impl->span())) [[unlikely]] {
+                throwOutOfMemoryError(globalObject, scope);
+                return false;
+            }
+        }
         return true;
     }
     throwTypeError(globalObject, scope, "Expected an ArrayBuffer, ArrayBufferView, or string chunk"_s);
@@ -709,25 +727,39 @@ static WTF::String finishTextAccumulator(JSC::VM& vm, JSGlobalObject* globalObje
             return rope.substring(1);
         return rope;
     }
-    WTF::Vector<uint8_t> bytes;
+    // estimatedLength never overcounts the bytes, so an estimate past the limit is final.
     const double estimatedLength = accumulator.estimatedLength;
-    if (estimatedLength > 0 && estimatedLength < static_cast<double>(std::numeric_limits<uint32_t>::max()))
-        bytes.reserveInitialCapacity(static_cast<size_t>(estimatedLength));
+    if (estimatedLength > static_cast<double>(WTF::StringImpl::MaxLength)
+        || exceedsStringLimit(static_cast<size_t>(estimatedLength))) [[unlikely]] {
+        releaseAccumulated();
+        throwOutOfMemoryError(globalObject, scope);
+        return WTF::String();
+    }
+    WTF::Vector<uint8_t> bytes;
+    if (estimatedLength > 0 && !bytes.tryReserveInitialCapacity(static_cast<size_t>(estimatedLength))) [[unlikely]] {
+        releaseAccumulated();
+        throwOutOfMemoryError(globalObject, scope);
+        return WTF::String();
+    }
     for (auto& piece : accumulator.pieces) {
         JSValue value = piece.get();
         if (!value)
             continue;
         bool appended = appendChunkBytes(vm, globalObject, value, bytes);
-        RETURN_IF_EXCEPTION(scope, WTF::String());
-        if (!appended)
+        if (scope.exception() || !appended) [[unlikely]] {
+            releaseAccumulated();
             return WTF::String();
+        }
     }
     if (accumulator.rope.length()) {
         WTF::String rope = accumulator.rope.toString();
         if (rope[0] == 0xFEFF)
             rope = rope.substring(1);
-        WTF::CString utf8 = rope.utf8();
-        bytes.append(std::span<const uint8_t> { reinterpret_cast<const uint8_t*>(utf8.data()), utf8.length() });
+        if (!appendUTF8WithinStringLimit(rope, bytes)) [[unlikely]] {
+            releaseAccumulated();
+            throwOutOfMemoryError(globalObject, scope);
+            return WTF::String();
+        }
     }
     releaseAccumulated();
     if (exceedsStringLimit(bytes.size())) [[unlikely]] {

src/jsc/bindings/webcore/streams/JSDirectStreamController.cpp

@@ -271,29 +271,46 @@ static String finishTextSink(JSC::VM& vm, JSGlobalObject* globalObject, JSDirect
         return rope;
     }
 
+    // Sizes are recorded at write() time; rejecting even if buffers were detached later is intended.
+    const double estimatedLength = accumulator.estimatedLength;
+    if (estimatedLength > static_cast<double>(WTF::StringImpl::MaxLength)
+        || Bun::WebStreams::exceedsStringLimit(static_cast<size_t>(estimatedLength))) [[unlikely]] {
+        throwOutOfMemoryError(globalObject, scope);
+        return String();
+    }
     Vector<uint8_t> bytes;
+    if (estimatedLength > 0 && !bytes.tryReserveInitialCapacity(static_cast<size_t>(estimatedLength))) [[unlikely]] {
+        throwOutOfMemoryError(globalObject, scope);
+        return String();
+    }
     for (auto& piece : accumulator.pieces) {
         JSValue value = piece.get();
+        bool appended = true;
         if (value.isString()) {
             String string = asString(value)->value(globalObject);
             RETURN_IF_EXCEPTION(scope, {});
-            auto utf8 = string.utf8();
-            bytes.append(std::span { reinterpret_cast<const uint8_t*>(utf8.data()), utf8.length() });
+            appended = Bun::WebStreams::appendUTF8WithinStringLimit(string, bytes);
         } else if (auto* view = dynamicDowncast<JSArrayBufferView>(value)) {
             if (!view->isDetached())
-                bytes.append(view->span());
+                appended = bytes.tryAppend(view->span());
         } else if (auto* buffer = dynamicDowncast<JSArrayBuffer>(value)) {
             auto* impl = buffer->impl();
             if (impl && !impl->isDetached())
-                bytes.append(impl->span());
+                appended = bytes.tryAppend(impl->span());
+        }
+        if (!appended) [[unlikely]] {
+            throwOutOfMemoryError(globalObject, scope);
+            return String();
         }
     }
     if (!accumulator.rope.isEmpty()) {
         String rope = accumulator.rope.toString();
         if (rope[0] == 0xFEFF)
             rope = rope.substring(1);
-        auto utf8 = rope.utf8();
-        bytes.append(std::span { reinterpret_cast<const uint8_t*>(utf8.data()), utf8.length() });
+        if (!Bun::WebStreams::appendUTF8WithinStringLimit(rope, bytes)) [[unlikely]] {
+            throwOutOfMemoryError(globalObject, scope);
+            return String();
+        }
     }
     if (Bun::WebStreams::exceedsStringLimit(bytes.size())) [[unlikely]] {
         throwOutOfMemoryError(globalObject, scope);

src/jsc/bindings/webcore/streams/WebStreamsInternals.h

@@ -597,6 +597,8 @@ JSC::JSValue readableStreamIntoText(JSC::JSGlobalObject*, JSReadableStream*); //
 JSC::JSValue readableStreamIntoArray(JSC::JSGlobalObject*, JSReadableStream*); // userJS: yes — BunStreamConsumers.cpp
 // Drop ONE leading U+FEFF, and only on the generic toText path.
 WTF::String withoutUTF8BOM(const WTF::String&); // userJS: no — BunStreamConsumers.cpp
+// Appends `string` UTF-8 encoded (lone surrogates become U+FFFD); false = over the string limit or allocation failed.
+bool appendUTF8WithinStringLimit(const WTF::String&, WTF::Vector<uint8_t>& bytes); // userJS: no — BunStreamConsumers.cpp
 
 // The three *Direct conversion paths.
 JSC::JSValue readableStreamToTextDirect(JSC::JSGlobalObject*, JSReadableStream*); // userJS: yes — BunStreamConsumers.cpp

test/js/web/streams/streams-string-limit.test.ts

@@ -0,0 +1,155 @@
+import { describe, expect, test } from "bun:test";
+import { bunEnv, bunExe } from "harness";
+import { totalmem } from "node:os";
+
+// Consuming a stream as text must reject with a catchable error when the accumulated
+// chunks exceed the maximum string length (2^31-1 bytes), instead of aborting the
+// process in WTF::Vector's capacity check. Each child commits ~2.2GB.
+const enoughMemory = totalmem() >= 8 * 1024 * 1024 * 1024;
+
+// 3 chunks of n bytes sum to 2^31+1: each chunk fits comfortably, the total does not.
+function consumeToText(streamSource: string): string {
+  return `
+    const n = 715827883;
+    const rs = ${streamSource};
+    try {
+      const text = await Bun.readableStreamToText(rs);
+      console.log("resolved", text.length);
+    } catch (e) {
+      console.log("threw", e.name, e.message);
+    }
+  `;
+}
+
+async function run(script: string): Promise<{ stdout: string; stderr: string; exitCode: number }> {
+  await using proc = Bun.spawn({
+    cmd: [bunExe(), "-e", script],
+    env: bunEnv,
+    stdout: "pipe",
+    stderr: "pipe",
+  });
+  const [stdout, stderr, exitCode] = await Promise.all([proc.stdout.text(), proc.stderr.text(), proc.exited]);
+  return { stdout, stderr, exitCode };
+}
+
+const threw = { stdout: "threw RangeError Out of memory\n", stderr: "", exitCode: 0 };
+
+describe.skipIf(!enoughMemory)("text consumers reject binary chunks summing past 2^31-1", () => {
+  test("queue-backed ReadableStream", async () => {
+    const result = await run(
+      consumeToText(`new ReadableStream({
+        start(c) {
+          for (let i = 0; i < 3; i++) c.enqueue(new Uint8Array(n));
+          c.close();
+        },
+      })`),
+    );
+    expect(result).toEqual(threw);
+  });
+
+  test("direct ReadableStream", async () => {
+    const result = await run(
+      consumeToText(`new ReadableStream({
+        type: "direct",
+        pull(c) {
+          for (let i = 0; i < 3; i++) c.write(new Uint8Array(n));
+          c.end();
+        },
+      })`),
+    );
+    expect(result).toEqual(threw);
+  });
+
+  test("Response(stream).text()", async () => {
+    const result = await run(`
+      const n = 715827883;
+      const rs = new ReadableStream({
+        start(c) {
+          for (let i = 0; i < 3; i++) c.enqueue(new Uint8Array(n));
+          c.close();
+        },
+      });
+      try {
+        const text = await new Response(rs).text();
+        console.log("resolved", text.length);
+      } catch (e) {
+        console.log("threw", e.name, e.message);
+      }
+    `);
+    expect(result).toEqual(threw);
+  });
+
+  // WTF::String::utf8() aborts once its conversion needs more than INT32_MAX bytes of
+  // scratch (2x the length for 8-bit strings), so a mixed stream with a near-limit string
+  // chunk crashed in the encode even when the real UTF-8 total fit. The consumers now size
+  // and write through simdutf instead.
+  test("mixed chunks with a big ASCII string chunk resolve when the UTF-8 total fits", async () => {
+    const result = await run(`
+        const big = Buffer.alloc(1200000000, "a").toString();
+        const rs = new ReadableStream({
+          start(c) {
+            c.enqueue(new Uint8Array([65]));
+            c.enqueue(big);
+            c.close();
+          },
+        });
+        try {
+          const text = await Bun.readableStreamToText(rs);
+          console.log("resolved", text.length);
+        } catch (e) {
+          console.log("threw", e.name, e.message);
+        }
+      `);
+    expect(result).toEqual({ stdout: "resolved 1200000001\n", stderr: "", exitCode: 0 });
+  }, 60_000);
+
+  test("mixed chunks whose UTF-8 expansion passes the limit reject", async () => {
+    const result = await run(`
+        // 1.2e9 U+00E9 chars: a Latin1 string whose UTF-8 form is 2.4e9 bytes.
+        const big = Buffer.alloc(1200000000, 233).toString("latin1");
+        const rs = new ReadableStream({
+          start(c) {
+            c.enqueue(new Uint8Array([65]));
+            c.enqueue(big);
+            c.close();
+          },
+        });
+        try {
+          const text = await Bun.readableStreamToText(rs);
+          console.log("resolved", text.length);
+        } catch (e) {
+          console.log("threw", e.name, e.message);
+        }
+      `);
+    expect(result).toEqual(threw);
+  }, 60_000);
+
+  // The direct sink records sizes at write() time and reads the spans at end(), so a
+  // resizable ArrayBuffer grown in between bypasses the up-front estimate check; the
+  // append itself must reject the oversized span.
+  test("direct stream with a buffer grown past the limit after write() rejects", async () => {
+    const result = await run(`
+      const ab = new ArrayBuffer(8, { maxByteLength: 2400000000 });
+      const rs = new ReadableStream({
+        type: "direct",
+        pull(c) {
+          c.write(new Uint8Array(ab));
+          ab.resize(2400000000);
+          console.log("resized", ab.byteLength);
+          c.end();
+        },
+      });
+      try {
+        const text = await Bun.readableStreamToText(rs);
+        console.log("resolved", text.length);
+      } catch (e) {
+        console.log("threw", e.name, e.message);
+      }
+    `);
+    expect(result).toEqual({
+      stdout: "resized 2400000000\nthrew RangeError Out of memory\n",
+      stderr: "",
+      exitCode: 0,
+    });
+  });
+});