diff --git a/src/js/internal/streams/native-readable.ts b/src/js/internal/streams/native-readable.ts index 65307212b821..583d6fc1b949 100644 --- a/src/js/internal/streams/native-readable.ts +++ b/src/js/internal/streams/native-readable.ts @@ -142,6 +142,9 @@ function read(this: NativeReadable, maxToRead: number) { if (typeof result === "number" && result > 1) { this[kHasResized] = true; this[kHighWaterMark] = Math.min(this[kHighWaterMark], result); + } else if (typeof result === "number" && result < 0) { + // Start::ReadyOwned: don't grow the pull view past Readable's hwm. + this[kHasResized] = true; } if ($isTypedArrayView(result) && result.byteLength > 0) { pushAndCheck(this, result); diff --git a/src/jsc/bindings/webcore/streams/BunStreamSource.cpp b/src/jsc/bindings/webcore/streams/BunStreamSource.cpp index 5784f4cdc03a..a567ee5342b7 100644 --- a/src/jsc/bindings/webcore/streams/BunStreamSource.cpp +++ b/src/jsc/bindings/webcore/streams/BunStreamSource.cpp @@ -417,6 +417,8 @@ static JSC::JSUint8Array* uint8Subarray(JSGlobalObject* globalObject, JSC::JSUin static JSC::JSUint8Array* nativeGetInternalBuffer(JSC::VM& vm, JSGlobalObject* globalObject, JSNativeStreamSourceAdapter* adapter) { auto scope = DECLARE_THROW_SCOPE(vm); + if (adapter->m_sourceOwnsChunks) + return nullptr; const size_t chunkSize = adapter->m_chunkSize; if (JSObject* pending = adapter->pendingView()) { auto* view = uncheckedDowncast(pending); @@ -478,8 +480,6 @@ static JSValue nativeDecodePullResult(JSC::VM& vm, JSGlobalObject* globalObject, return jsUndefined(); } if (auto* chunk = dynamicDowncast(result)) { - if (!isClosed) - nativeAdjustChunkSize(adapter, chunk->byteLength()); if (chunk->byteLength() > 0) { if (adapter->m_textMode) { nativeEnqueueTextChunk(globalObject, controller, adapter->m_textState, chunk->span(), /* flush */ false); @@ -556,7 +556,13 @@ void materializeNativeSource(JSGlobalObject* globalObject, JSReadableStream* str auto* adapter = WebCore::JSNativeStreamSourceAdapter::create(vm, runtime->nativeStreamSourceAdapterStructure(domGlobalObject)); adapter->setHandle(vm, handle); adapter->m_textMode = stream->m_nativeTextMode; - adapter->m_chunkSize = std::max(static_cast(chunkSize), autoAllocateChunkSize); + if (chunkSize < 0) { + adapter->m_sourceOwnsChunks = true; + adapter->m_hasResized = true; + adapter->m_chunkSize = 0; + } else { + adapter->m_chunkSize = std::max(static_cast(chunkSize), autoAllocateChunkSize); + } auto* closer = JSC::constructEmptyArray(globalObject, nullptr, 1); RETURN_IF_EXCEPTION(scope, ); closer->putDirectIndex(globalObject, 0, jsBoolean(false)); @@ -645,7 +651,7 @@ static JSPromise* nativeSourcePullImpl(JSC::VM& vm, JSGlobalObject* globalObject RETURN_IF_EXCEPTION(scope, nullptr); MarkedArgumentBuffer pullArgs; - pullArgs.append(view); + pullArgs.append(view ? JSValue(view) : jsUndefined()); pullArgs.append(closer); ASSERT(!pullArgs.hasOverflowed()); JSValue result = invokeMethod(vm, globalObject, handle, builtinNames(vm).pullPublicName(), pullArgs); diff --git a/src/jsc/bindings/webcore/streams/BunStreamSource.h b/src/jsc/bindings/webcore/streams/BunStreamSource.h index 95dd4bebdeb6..5b201a6c1174 100644 --- a/src/jsc/bindings/webcore/streams/BunStreamSource.h +++ b/src/jsc/bindings/webcore/streams/BunStreamSource.h @@ -88,6 +88,8 @@ class JSNativeStreamSourceAdapter final : public JSC::JSInternalFieldObjectImpl< bool m_closed : 1 { false }; // Body.textStream(): each pulled byte span is UTF-8-decoded before enqueue. bool m_textMode : 1 { false }; + // start() returned <0 (Start::ReadyOwned): never allocate PendingView. + bool m_sourceOwnsChunks : 1 { false }; Bun::WebStreams::StreamingUTF8DecodeState m_textState; private: diff --git a/src/runtime/webcore/Body.rs b/src/runtime/webcore/Body.rs index 201e73b7c338..ff8b78c8736c 100644 --- a/src/runtime/webcore/Body.rs +++ b/src/runtime/webcore/Body.rs @@ -884,7 +884,6 @@ impl Value { match drain_result { DrainResult::EstimatedSize(estimated_size) => { - reader.context.high_water_mark = estimated_size as blob::SizeType; reader .context .size_hint @@ -1509,7 +1508,6 @@ impl Value { match drain_result { DrainResult::EstimatedSize(estimated_size) => { - reader.context.high_water_mark = estimated_size as blob::SizeType; reader .context .size_hint diff --git a/src/runtime/webcore/ByteStream.rs b/src/runtime/webcore/ByteStream.rs index f873f3b231b0..ae782c80078b 100644 --- a/src/runtime/webcore/ByteStream.rs +++ b/src/runtime/webcore/ByteStream.rs @@ -14,9 +14,7 @@ bun_output::declare_scope!(ByteStream, visible); /// R-2 (`sharedThis`): every JS-reachable inherent method takes `&self` so a /// re-entrant JS call (e.g. `pending.run()` → JS → `onPull`) cannot stack two /// `&mut ByteStream`. Fields mutated on those paths are wrapped in `Cell` -/// (Copy scalars / raw ptrs) or [`JsCell`] (non-Copy). `high_water_mark` / -/// `size_hint` are written only at init time (before the JS wrapper exists) -/// and stay bare. +/// (Copy scalars / raw ptrs) or [`JsCell`] (non-Copy). /// /// The `SourceContext` trait still spells its callbacks `&mut self` (shared /// across `ByteBlobLoader` / `FileReader`); the trait impl below auto-derefs @@ -32,7 +30,6 @@ pub struct ByteStream { pub(crate) pending_buffer: Cell<*mut [u8]>, pub(crate) pending_value: JsCell, // jsc.Strong.Optional pub offset: Cell, - pub(crate) high_water_mark: blob::SizeType, /// Native sink this stream is piped into; `on_data` dispatches and honors `Writable`. pub(crate) sink: JsCell, /// Set on `Writable::Backpressure` (buffer instead of write); cleared by [`Self::resume`]. @@ -54,7 +51,6 @@ impl Default for ByteStream { pending_buffer: Cell::new(Self::empty_pending_buffer()), pending_value: JsCell::new(StrongOptional::empty()), offset: Cell::new(0), - high_water_mark: 0, sink: JsCell::new(SinkHandle::None), sink_paused: Cell::new(false), size_hint: Cell::new(0), @@ -133,17 +129,7 @@ impl ByteStream { return streams::Start::OwnedAndDone(Vec::::move_from_list(buffer)); } - if self.high_water_mark == 0 { - return streams::Start::Ready; - } - - // For HTTP, the maximum streaming response body size will be 512 KB. - // #define LIBUS_RECV_BUFFER_LENGTH 524288 - // For HTTPS, the size is probably quite a bit lower like 64 KB due to TLS transmission. - // We add 1 extra page size so that if there's a little bit of excess buffered data, we avoid extra allocations. - let page_size: blob::SizeType = - blob::SizeType::try_from(bun_sys::page_size()).expect("int cast"); - streams::Start::ChunkSize((512 * 1024 + page_size).min(self.high_water_mark.max(page_size))) + streams::Start::ReadyOwned } fn value(&self) -> JSValue { @@ -376,10 +362,55 @@ impl ByteStream { return Ok(()); } - let chunk = stream.slice(); - if self.pending.get().state == streams::PendingState::Pending { debug_assert!(self.buffer.get().is_empty()); + + // Pending pull with no view parked (C++ adapter): Owned handoff. + if self.pending_value.get().get().is_none() { + let is_done = self.has_received_last_chunk.get(); + let result = match stream { + streams::Result::Err(_) => { + self.done.set(true); + stream + } + streams::Result::Done => { + self.done.set(true); + streams::Result::Done + } + streams::Result::Owned(owned) | streams::Result::OwnedAndDone(owned) => { + if is_done { + self.done.set(true); + if owned.is_empty() { + streams::Result::Done + } else { + streams::Result::OwnedAndDone(owned) + } + } else { + streams::Result::Owned(owned) + } + } + streams::Result::Temporary(temp) | streams::Result::TemporaryAndDone(temp) => { + let owned = temp.slice().to_vec(); + if is_done { + self.done.set(true); + if owned.is_empty() { + streams::Result::Done + } else { + streams::Result::OwnedAndDone(owned) + } + } else { + streams::Result::Owned(owned) + } + } + _ => unreachable!(), + }; + self.pending.with_mut(|p| p.result = result); + self.signal_drained(); + self.pending.with_mut(|p| p.run()); + return Ok(()); + } + + let chunk = stream.slice(); // Re-derive the destination from the GC-rooted view instead of trusting the // raw pointer captured at pull time: JS can detach or transfer the backing // ArrayBuffer between the pull and the data arriving, leaving @@ -525,9 +556,32 @@ impl ByteStream { fn on_pull(&self, buffer: &mut [u8], view: JSValue) -> streams::Result { bun_jsc::mark_binding!(); - debug_assert!(!buffer.is_empty()); debug_assert!(self.buffer_action.get().is_none()); + // No pull view = C++ adapter Owned handoff; view = native-readable's metered copy below. + if buffer.is_empty() { + if !self.buffer.get().is_empty() { + debug_assert!(self.pending_value.get().get().is_none()); + debug_assert_eq!(self.offset.get(), 0); + let owned = self.buffer.replace(Vec::new()); + self.signal_drained(); + if self.has_received_last_chunk.get() { + self.done.set(true); + return streams::Result::OwnedAndDone(owned); + } + return streams::Result::Owned(owned); + } + if self.has_received_last_chunk.get() { + if matches!(self.pending.get().result, streams::Result::Err(_)) { + return self + .pending + .with_mut(|p| core::mem::replace(&mut p.result, streams::Result::Done)); + } + return streams::Result::Done; + } + return streams::Result::Pending(self.pending.as_ptr()); + } + if !self.buffer.get().is_empty() { debug_assert!(self.value().is_empty()); // == .zero // R-2: confine the `&mut Vec` to a `with_mut` so no `JsCell` @@ -592,7 +646,6 @@ impl ByteStream { fn on_cancel(&self) { bun_jsc::mark_binding!(); - let view = self.value(); if self.buffer.get().capacity() > 0 { self.buffer.with_mut(|b| { b.clear(); @@ -602,7 +655,7 @@ impl ByteStream { self.done.set(true); self.pending_value.with_mut(|pv| pv.deinit()); - if !view.is_empty() { + if self.pending.get().state == streams::PendingState::Pending { self.pending_buffer.set(Self::empty_pending_buffer()); self.pending.with_mut(|p| { p.result.release(); diff --git a/src/runtime/webcore/ReadableStream.rs b/src/runtime/webcore/ReadableStream.rs index 6ef9d5771301..342c0e0ad7c4 100644 --- a/src/runtime/webcore/ReadableStream.rs +++ b/src/runtime/webcore/ReadableStream.rs @@ -1117,10 +1117,11 @@ impl NewSource { let this_jsvalue = call_frame.this(); let [view, flags] = call_frame.arguments_as_array::<2>(); view.ensure_still_alive(); - let Some(mut buffer) = view.as_array_buffer(global_this) else { - return Ok(JSValue::UNDEFINED); + let result = if let Some(mut buffer) = view.as_array_buffer(global_this) { + self.on_pull_from_js(buffer.slice_mut(), view) + } else { + self.on_pull_from_js(&mut [], view) }; - let result = self.on_pull_from_js(buffer.slice_mut(), view); Self::process_result(this_jsvalue, global_this, flags, result) } @@ -1133,6 +1134,7 @@ impl NewSource { match self.on_start_from_js() { streams::Start::Empty => Ok(JSValue::js_number(0.0)), streams::Start::Ready => Ok(JSValue::js_number(16384.0)), + streams::Start::ReadyOwned => Ok(JSValue::js_number(-1.0)), streams::Start::ChunkSize(size) => Ok(JSValue::js_number(size as f64)), streams::Start::Err(err) => Err(global_this.throw_value(err.to_js(global_this))), rc => rc.to_js(global_this), diff --git a/src/runtime/webcore/streams.rs b/src/runtime/webcore/streams.rs index a26a8e0305a0..b8138c45c72b 100644 --- a/src/runtime/webcore/streams.rs +++ b/src/runtime/webcore/streams.rs @@ -73,6 +73,8 @@ pub enum Start { }, FileSink(FileSinkOptions), Ready, + /// `on_pull` hands back `Owned` chunks; the adapter skips its pull view. + ReadyOwned, OwnedAndDone(Vec), } @@ -97,7 +99,7 @@ pub enum StartTag { impl Start { pub fn to_js(self, global_this: &JSGlobalObject) -> JsResult { match self { - Start::Empty | Start::Ready => Ok(JSValue::UNDEFINED), + Start::Empty | Start::Ready | Start::ReadyOwned => Ok(JSValue::UNDEFINED), Start::ChunkSize(chunk) => Ok(JSValue::from(chunk)), Start::Err(err) => Err(err.throw(global_this)), Start::OwnedAndDone(list) => { diff --git a/test/js/bun/http/serve-request-body-pipeline-memory.test.ts b/test/js/bun/http/serve-request-body-pipeline-memory.test.ts new file mode 100644 index 000000000000..792f7fb33f65 --- /dev/null +++ b/test/js/bun/http/serve-request-body-pipeline-memory.test.ts @@ -0,0 +1,114 @@ +// `for await (chunk of req.body)` (what `pipeline(req.body, writable)` runs +// via `pumpToNode`) should yield right-sized chunks. A `Bun.serve` request +// body is a push source (`ByteStream`): bytes arrive via `on_data` and are +// handed to the reader as the owning allocation. Previously `on_pull`/`on_data` +// copied into the native-source adapter's ~256-512 KiB scratch view and +// surfaced each chunk as a subarray over that whole backing, so every in-flight +// request held a ~0.5 MB `ArrayBuffer` regardless of how little data was +// actually read. + +import { expect, test } from "bun:test"; +import { connect } from "node:net"; + +type Seen = { len: number; backing: number; off: number }; + +async function runUpload( + handler: (req: Request, onParked: () => void, onChunk: () => void) => Promise, + bodyBytes: number, + writes: number[], +): Promise { + const { promise: handlerP, resolve: handlerDone, reject: handlerFail } = Promise.withResolvers(); + const { promise: pullParkedP, resolve: pullParked, reject: pullParkedFail } = Promise.withResolvers(); + // One ack per client write so the next write only leaves after the server + // has observed the previous one as a distinct on_data chunk. + const acks = writes.map(() => Promise.withResolvers()); + // `fail` rejects every promise, but only the one currently awaited has a + // handler at that point; mark the rest observed so a regression surfaces as + // one failure instead of one plus N unhandled-rejection noise entries. + void handlerP.catch(() => {}); + void pullParkedP.catch(() => {}); + for (const a of acks) void a.promise.catch(() => {}); + let ackIndex = 0; + const onChunk = () => acks[ackIndex++]?.resolve(); + const fail = (e: unknown) => { + handlerFail(e); + pullParkedFail(e); + for (const a of acks) a.reject(e); + }; + + await using server = Bun.serve({ + port: 0, + async fetch(req) { + try { + const seen = await handler(req, () => queueMicrotask(() => queueMicrotask(pullParked)), onChunk); + handlerDone(seen); + } catch (e) { + fail(e); + } + return new Response("ok"); + }, + }); + + const sock = connect({ port: server.port, host: "127.0.0.1" }); + sock.on("error", fail); + await new Promise((res, rej) => { + sock.once("connect", () => res()); + sock.once("error", rej); + }); + + try { + sock.write(`POST / HTTP/1.1\r\nHost: x\r\nContent-Length: ${bodyBytes}\r\nConnection: close\r\n\r\n`); + await pullParkedP; + for (const [i, n] of writes.entries()) { + sock.write(Buffer.alloc(n, 0x61)); + await acks[i].promise; + } + sock.end(); + return await handlerP; + } finally { + sock.destroy(); + } +} + +function checkRightSized(seen: Seen[], expectedTotal: number, minChunks: number) { + let total = 0; + for (const { len } of seen) total += len; + expect(total).toBe(expectedTotal); + expect(seen.length).toBeGreaterThanOrEqual(minChunks); + // Every chunk is its own allocation. On main each chunk was a subarray into + // a single ~516 KiB (for await) / ~64 KiB (fromWeb) scratch view, so + // `backing >> len` and `off` advanced per chunk. + for (const { len, backing, off } of seen) { + expect({ len, backing, off }).toEqual({ len, backing: len, off: 0 }); + } +} + +test("for await (req.body) chunks are backed by right-sized buffers, not the adapter's scratch view", async () => { + // Content-Length large enough that on_start would have sized the pull view + // at its ~512 KiB ceiling on main. + const BODY = 2 * 1024 * 1024; + const writes = [8 * 1024, 8 * 1024, BODY - 16 * 1024]; + const seen = await runUpload( + async (req, onParked, onChunk) => { + const out: Seen[] = []; + // Let the client know the first pull has parked (no body bytes yet) so + // the first chunk is resolved from on_data, not from drain(). + onParked(); + for await (const chunk of req.body!) { + out.push({ len: chunk.byteLength, backing: chunk.buffer.byteLength, off: chunk.byteOffset }); + onChunk(); + } + return out; + }, + BODY, + writes, + ); + checkRightSized(seen, BODY, writes.length); +}); + +// `Readable.fromWeb(req.body)` is deliberately left on the copy-into-view path: +// node:stream Readable calls `_read` ahead of downstream consumption, so +// metering via the pull view's size is what keeps the producer paused while +// a chunk is still sitting in the Readable's own buffer. The C++ adapter +// above pulls once per reader.read(), so it can hand off whole buffers +// without that hazard. diff --git a/test/js/web/streams/streams-leak.test.ts b/test/js/web/streams/streams-leak.test.ts index 0a2862ade9c1..4308d849fc7d 100644 --- a/test/js/web/streams/streams-leak.test.ts +++ b/test/js/web/streams/streams-leak.test.ts @@ -1,7 +1,7 @@ import { expect, test } from "bun:test"; import { bunEnv, bunExe, isASAN, isWindows, rss, tempDir } from "harness"; -test("native ReadableStream reuses the pull buffer across small reads", async () => { +test("native ReadableStream backs small reads with right-sized buffers", async () => { // #getInternalBuffer used to rotate to a fresh autoAllocateChunkSize // (256KB) Uint8Array whenever $data.length < chunkSize — true after // every nonzero read, since #handleNumberResult stores the tail @@ -60,14 +60,13 @@ test("native ReadableStream reuses the pull buffer across small reads", async () // through the native pull path. expect(chunks.length).toBeGreaterThanOrEqual(CHUNKS_TO_WRITE); - // Consecutive small reads should land in the same backing buffer (the - // tail subarray is reused until a read fills it). 128 bytes of 2-byte - // chunks fits well inside one 256KB buffer, so the whole stream should - // share a handful at most. Pre-fix every chunk had its own 256KB - // buffer, so this was ~chunks.length. - const distinctBuffers = new Set(chunks.map(c => c.buffer)); - expect(distinctBuffers.size).toBeLessThan(8); + // Each chunk is handed off as its own right-sized allocation; pre-fix each + // was a subarray over a fresh 256 KB Gigacage buffer. + for (const c of chunks) { + expect(c.buffer.byteLength).toBe(c.byteLength); + } + const distinctBuffers = new Set(chunks.map(c => c.buffer)); let backingBytes = 0; for (const buf of distinctBuffers) backingBytes += buf.byteLength; // Pre-fix this was ~chunks.length * 256KB ≈ 16 MB.