Stop aborting when stream arrayBuffer/bytes consumers mix chunk types past 2 GiB

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

@@ -357,12 +357,10 @@ static JSValue concatenateChunks(JSC::VM& vm, JSGlobalObject* globalObject, JSAr
     auto scope = DECLARE_THROW_SCOPE(vm);
     unsigned length = chunks->length();
 
-    // ONE pass over the array: read each element exactly once, materialize each string
-    // exactly once, and size the output as we go. `values` roots every chunk across the
-    // string materializations; `stringChunks` carries each string and its UTF-8 size so
-    // the write pass below never re-reads the array or re-encodes.
+    // One pass: read and measure each chunk exactly once (strings materialize into
+    // `measuredChunks`) so the write pass never re-reads or re-encodes; `values` roots the chunks.
     MarkedArgumentBuffer values;
-    WTF::Vector<std::pair<WTF::String, size_t>, 16> stringChunks;
+    WTF::Vector<std::pair<WTF::String, size_t>, 16> measuredChunks;
     bool anyString = false;
     WTF::CheckedSize total = 0;
     for (unsigned i = 0; i < length; i++) {
@@ -375,19 +373,21 @@ static JSValue concatenateChunks(JSC::VM& vm, JSGlobalObject* globalObject, JSAr
             RETURN_IF_EXCEPTION(scope, {});
             size_t byteLength = utf8ByteLengthWithReplacement(string);
             total += byteLength;
-            stringChunks.append({ WTF::move(string), byteLength });
+            measuredChunks.append({ WTF::move(string), byteLength });
             continue;
         }
-        stringChunks.append({ WTF::String(), 0 });
+        size_t byteLength = 0;
         if (auto* view = dynamicDowncast<JSC::JSArrayBufferView>(chunk))
-            total += view->isDetached() ? 0 : view->byteLength();
+            byteLength = view->isDetached() ? 0 : view->byteLength();
         else if (auto* jsBuffer = dynamicDowncast<JSC::JSArrayBuffer>(chunk)) {
             auto* impl = jsBuffer->impl();
-            total += (impl && !impl->isDetached()) ? impl->byteLength() : 0;
+            byteLength = (impl && !impl->isDetached()) ? impl->byteLength() : 0;
         } else {
             throwTypeError(globalObject, scope, "Expected an ArrayBuffer, ArrayBufferView, or string chunk"_s);
             return {};
         }
+        total += byteLength;
+        measuredChunks.append({ WTF::String(), byteLength });
     }
     if (values.hasOverflowed()) [[unlikely]] {
         throwOutOfMemoryError(globalObject, scope);
@@ -402,47 +402,60 @@ static JSValue concatenateChunks(JSC::VM& vm, JSGlobalObject* globalObject, JSAr
         throwOutOfMemoryError(globalObject, scope);
         return {};
     }
-    WTF::Vector<uint8_t> bytes;
-    bytes.reserveInitialCapacity(total.value());
+    // Assemble directly into the result ArrayBuffer: WTF::Vector<uint8_t> CRASH()es past
+    // its INT32_MAX capacity cap; ArrayBuffer capacities are size_t and fail by throwing.
+    RefPtr<JSC::ArrayBuffer> resultBuffer = JSC::ArrayBuffer::tryCreateUninitialized(total.value(), 1);
+    if (!resultBuffer) [[unlikely]] {
+        throwOutOfMemoryError(globalObject, scope);
+        return {};
+    }
+    std::span<uint8_t> bytes { static_cast<uint8_t*>(resultBuffer->data()), total.value() };
+    size_t offset = 0;
     for (unsigned i = 0; i < length; i++) {
-        auto& [string, stringByteLength] = stringChunks[i];
+        auto& [string, measuredByteLength] = measuredChunks[i];
         if (!string.isNull()) {
-            if (stringByteLength) {
-                size_t oldSize = bytes.size();
-                bytes.grow(oldSize + stringByteLength);
-                size_t written = writeUTF8(string, bytes.mutableSpan().subspan(oldSize));
-                // The sizer and writer must agree; never expose ungrown (uninitialized) bytes.
-                ASSERT(written == stringByteLength);
-                if (written < stringByteLength) [[unlikely]]
-                    bytes.shrink(oldSize + written);
+            if (measuredByteLength) {
+                size_t written = writeUTF8(string, bytes.subspan(offset, measuredByteLength));
+                // The sizer/writer pair agrees by contract; the writer reports its count through
+                // a u32 (0 for an exactly-2^32-byte string), so advance by the measured size.
+                ASSERT_UNUSED(written, written == measuredByteLength || measuredByteLength == (1ull << 32));
+                offset += measuredByteLength;
             }
             continue;
         }
         JSValue chunk = values.at(i);
+        std::span<const uint8_t> span;
         if (auto* view = dynamicDowncast<JSC::JSArrayBufferView>(chunk)) {
             if (!view->isDetached())
-                bytes.append(view->span());
+                span = view->span();
         } else if (auto* jsBuffer = dynamicDowncast<JSC::JSArrayBuffer>(chunk)) {
             if (auto* impl = jsBuffer->impl(); impl && !impl->isDetached())
-                bytes.append(impl->span());
+                span = impl->span();
         }
+        // Clamp to the measured size: a chunk grown since the sizing pass (getter reentry
+        // during getIndex above, or a cross-thread SharedArrayBuffer grow) must not overrun.
+        size_t copyLength = std::min(span.size(), measuredByteLength);
+        if (copyLength)
+            memcpy(bytes.data() + offset, span.data(), copyLength);
+        offset += copyLength;
     }
-    if (asUint8Array) {
-        // Buffer-backed from birth: a later `.buffer` access never has to change modes.
-        RefPtr<JSC::ArrayBuffer> resultBuffer = JSC::ArrayBuffer::tryCreate(bytes.span());
-        if (!resultBuffer) [[unlikely]] {
+    if (offset < bytes.size()) [[unlikely]] {
+        // A chunk shrank or detached after measurement: re-create at the actual size
+        // instead of exposing the uninitialized tail.
+        RefPtr<JSC::ArrayBuffer> rightSized = JSC::ArrayBuffer::tryCreate(bytes.first(offset));
+        if (!rightSized) [[unlikely]] {
             throwOutOfMemoryError(globalObject, scope);
             return {};
         }
-        auto* structure = globalObject->typedArrayStructureWithTypedArrayType<JSC::TypeUint8>();
-        RELEASE_AND_RETURN(scope, JSC::JSUint8Array::create(globalObject, structure, WTF::move(resultBuffer), 0, bytes.size()));
+        resultBuffer = WTF::move(rightSized);
     }
-    auto buffer = JSC::ArrayBuffer::tryCreate(bytes.span());
-    if (!buffer) [[unlikely]] {
-        throwOutOfMemoryError(globalObject, scope);
-        return {};
+    size_t byteLength = resultBuffer->byteLength();
+    if (asUint8Array) {
+        // Buffer-backed from birth: a later `.buffer` access never has to change modes.
+        auto* structure = globalObject->typedArrayStructureWithTypedArrayType<JSC::TypeUint8>();
+        RELEASE_AND_RETURN(scope, JSC::JSUint8Array::create(globalObject, structure, WTF::move(resultBuffer), 0, byteLength));
     }
-    return JSC::JSArrayBuffer::create(vm, globalObject->arrayBufferStructure(JSC::ArrayBufferSharingMode::Default), WTF::move(buffer));
+    return JSC::JSArrayBuffer::create(vm, globalObject->arrayBufferStructure(JSC::ArrayBufferSharingMode::Default), WTF::move(resultBuffer));
 }
 
 static bool binaryChunkSpan(JSValue chunk, std::span<const uint8_t>& out)

test/js/bun/util/readablestreamtoarraybuffer.test.ts

@@ -1,4 +1,10 @@
 import { expect, test } from "bun:test";
+import { bunEnv, bunExe } from "harness";
+import os from "node:os";
+
+// Same gate as blob-oom.test.ts / buffer.test.js 4 GiB cases: the multi-GiB tests below
+// need real memory (Windows commits allocations upfront; Linux would OOM-kill, not throw).
+const hasEnoughMemory = os.totalmem() >= 10 * 1024 ** 3;
 
 // The consumer's own promise plumbing must never route through user-patched
 // Promise.prototype.then. (A thenable returned by the user's own start() is
@@ -28,6 +34,245 @@ test("readableStreamToArrayBuffer does not call a patched Promise.prototype.then
   }
 });
 
+// Mixed string+binary chunk arrays used to be concatenated through a WTF::Vector, whose
+// INT32_MAX capacity cap aborts the whole process once the total crosses 2 GiB. The
+// result must instead be assembled in an ArrayBuffer (fine up to 4 GiB on 64-bit), and
+// totals past the 4 GiB ArrayBuffer maximum must throw a catchable error.
+// Subprocess tests: the failure mode is a process abort, and the large allocations
+// should not live in the test runner. Each script prints SKIP if this machine cannot
+// allocate the inputs.
+for (const consumer of ["readableStreamToArrayBuffer", "readableStreamToBytes"]) {
+  test.skipIf(!hasEnoughMemory)(`${consumer} handles mixed string+binary chunks totaling over 2 GiB`, async () => {
+    // The trailing "!?" makes a string write land at an offset past 2^31, not just past 2^30.
+    const script = `
+      const GIB = 1073741824;
+      let a, b;
+      try {
+        a = new Uint8Array(GIB);
+        b = new ArrayBuffer(GIB);
+      } catch {
+        console.log("SKIP");
+        process.exit(0);
+      }
+      a[0] = 0xaa;
+      a[GIB - 1] = 0xab;
+      const bView = new Uint8Array(b);
+      bView[0] = 0xba;
+      bView[GIB - 1] = 0xbb;
+      const stream = new ReadableStream({
+        start(c) {
+          c.enqueue(a);
+          c.enqueue("xyz");
+          c.enqueue(b);
+          c.enqueue("!?");
+          c.close();
+        },
+      });
+      const result = await Bun.${consumer}(stream);
+      const r = result instanceof Uint8Array ? result : new Uint8Array(result);
+      console.log(
+        JSON.stringify({
+          constructor: result.constructor.name,
+          byteLength: result.byteLength,
+          markers: [r[0], r[GIB - 1], r[GIB], r[GIB + 1], r[GIB + 2], r[GIB + 3], r[2 * GIB + 2], r[2 * GIB + 3], r[2 * GIB + 4]],
+        }),
+      );
+    `;
+    await using proc = Bun.spawn({
+      cmd: [bunExe(), "-e", script],
+      env: bunEnv,
+      stdout: "pipe",
+      stderr: "inherit",
+    });
+    const [stdout, exitCode] = await Promise.all([proc.stdout.text(), proc.exited]);
+    if (stdout.trim() === "SKIP") return;
+    // String comparison instead of JSON.parse: a crashed child diffs against "" here
+    // rather than failing with an unrelated parse error.
+    expect(stdout.trim()).toBe(
+      JSON.stringify({
+        constructor: consumer === "readableStreamToBytes" ? "Uint8Array" : "ArrayBuffer",
+        byteLength: 2 * 1073741824 + 5,
+        markers: [0xaa, 0xab, 0x78, 0x79, 0x7a, 0xba, 0xbb, 0x21, 0x3f],
+      }),
+    );
+    expect(proc.signalCode).toBe(null);
+    expect(exitCode).toBe(0);
+  });
+
+  test.skipIf(!hasEnoughMemory)(
+    `${consumer} throws instead of aborting when mixed chunks exceed the 4 GiB ArrayBuffer maximum`,
+    async () => {
+      // The inputs are never touched (zero pages on Linux; Windows still commits them,
+      // hence the memory gate) and the oversized result allocation fails fast.
+      const script = `
+      const n = 2200000000;
+      let a, b;
+      try {
+        a = new Uint8Array(n);
+        b = new Uint8Array(n);
+      } catch {
+        console.log("SKIP");
+        process.exit(0);
+      }
+      const stream = new ReadableStream({
+        start(c) {
+          c.enqueue(a);
+          c.enqueue(b);
+          c.enqueue("x");
+          c.close();
+        },
+      });
+      try {
+        await Bun.${consumer}(stream);
+        console.log("NO_THROW");
+      } catch (e) {
+        console.log("THREW RangeError=" + (e instanceof RangeError));
+      }
+    `;
+      await using proc = Bun.spawn({
+        cmd: [bunExe(), "-e", script],
+        env: bunEnv,
+        stdout: "pipe",
+        stderr: "inherit",
+      });
+      const [stdout, exitCode] = await Promise.all([proc.stdout.text(), proc.exited]);
+      if (stdout.trim() === "SKIP") return;
+      expect(stdout.trim()).toBe("THREW RangeError=true");
+      expect(proc.signalCode).toBe(null);
+      expect(exitCode).toBe(0);
+    },
+  );
+}
+
+// The string-encoding half of the mixed arm at scale: a single string chunk whose UTF-8
+// size exceeds 2^31. UTF-16 source (U+0808 is 3 UTF-8 bytes: E0 A0 88) so the conversion
+// takes the simdutf path; the Latin-1 writer is a scalar loop that takes minutes in debug.
+test.skipIf(!hasEnoughMemory)(
+  "readableStreamToBytes handles a string chunk larger than 2 GiB",
+  async () => {
+    const script = `
+      const n = 716000000; // 3 * n = 2148000000 > 2^31
+      let s, marker;
+      try {
+        s = Buffer.alloc(2 * n, 0x08).toString("utf16le");
+        marker = new Uint8Array([1, 2, 3]);
+      } catch {
+        console.log("SKIP");
+        process.exit(0);
+      }
+      const stream = new ReadableStream({
+        start(c) {
+          c.enqueue(s);
+          c.enqueue(marker);
+          c.close();
+        },
+      });
+      const r = await Bun.readableStreamToBytes(stream);
+      console.log(
+        JSON.stringify({
+          byteLength: r.byteLength,
+          markers: [r[0], r[1], r[2], r[3 * n - 3], r[3 * n - 2], r[3 * n - 1], r[3 * n], r[3 * n + 1], r[3 * n + 2]],
+        }),
+      );
+    `;
+    await using proc = Bun.spawn({
+      cmd: [bunExe(), "-e", script],
+      env: bunEnv,
+      stdout: "pipe",
+      stderr: "inherit",
+    });
+    const [stdout, exitCode] = await Promise.all([proc.stdout.text(), proc.exited]);
+    if (stdout.trim() === "SKIP") return;
+    expect(stdout.trim()).toBe(
+      JSON.stringify({
+        byteLength: 3 * 716000000 + 3,
+        markers: [0xe0, 0xa0, 0x88, 0xe0, 0xa0, 0x88, 1, 2, 3],
+      }),
+    );
+    expect(proc.signalCode).toBe(null);
+    expect(exitCode).toBe(0);
+  },
+  // ~18s under the debug+ASAN build: 3.5 GiB of string building plus a 2 GiB encode.
+  60_000,
+);
+
+// Array.prototype indexed accessors intercept the consumer's internal chunk-array pushes
+// (JSC havingABadTime), so a getter runs during the sizing pass and can resize, detach, or
+// grow an already-measured chunk before the write pass copies it. A grown chunk must be
+// clamped to its measured size (the result buffer is pre-sized; an unclamped copy would
+// overrun it) and a shrunken or detached chunk must yield a result of the actual bytes
+// written, never uninitialized memory.
+test("chunks mutated by Array.prototype accessor reentry during concatenation are clamped", async () => {
+  const script = `
+    let log = "";
+    let captured = {};
+    let mutate = () => {};
+    for (let i = 0; i < 2; i++) {
+      Object.defineProperty(Array.prototype, String(i), {
+        configurable: true,
+        set(v) {
+          log += "set" + i + ",";
+          captured[i] = v;
+        },
+        get() {
+          log += "get" + i + ",";
+          if (i === 1) mutate();
+          return captured[i];
+        },
+      });
+    }
+    // No array literals or pushes below (they would route through the accessors).
+    async function run(mutation) {
+      captured = {};
+      log = "";
+      const rab = new ArrayBuffer(64, { maxByteLength: 128 });
+      const view = new Uint8Array(rab);
+      view.fill(0x41);
+      mutate = () => {
+        if (mutation === "shrink") rab.resize(3);
+        else if (mutation === "transfer") rab.transfer();
+        else rab.resize(128);
+      };
+      let pulls = 0;
+      const stream = new ReadableStream(
+        {
+          async pull(c) {
+            pulls++;
+            if (pulls === 1) {
+              c.enqueue(view);
+            } else {
+              c.enqueue("x");
+              c.close();
+            }
+          },
+        },
+        { highWaterMark: 0 },
+      );
+      const r = new Uint8Array(await Bun.readableStreamToArrayBuffer(stream));
+      let allA = true;
+      for (let i = 0; i < r.length - 1; i++) allA = allA && r[i] === 0x41;
+      console.log(mutation + " " + log + " len=" + r.length + " allA=" + allA + " tail=" + r[r.length - 1]);
+    }
+    await run("shrink");
+    await run("transfer");
+    await run("grow");
+  `;
+  await using proc = Bun.spawn({
+    cmd: [bunExe(), "-e", script],
+    env: bunEnv,
+    stdout: "pipe",
+    stderr: "inherit",
+  });
+  const [stdout, exitCode] = await Promise.all([proc.stdout.text(), proc.exited]);
+  expect(stdout.trim().split("\n")).toEqual([
+    "shrink set0,set1,get0,get1, len=4 allA=true tail=120",
+    "transfer set0,set1,get0,get1, len=1 allA=true tail=120",
+    "grow set0,set1,get0,get1, len=65 allA=true tail=120",
+  ]);
+  expect(proc.signalCode).toBe(null);
+  expect(exitCode).toBe(0);
+});
+
 test("an async start() promise is adopted observably, like Node", async () => {
   const originalThen = Promise.prototype.then;
   let counter = 0;