streams: end a ReadableStream when the native sink it is locked to is done

Merged
RO
2mo ago
+269-45

Problem

  • A fetch() body or a Bun.file() stream stays readable after a native consumer reads it: a fetch() request body, Bun.write(), a Bun.serve() response body, Bun.spawn() stdin, an S3 upload, or HTMLRewriter. isReadable(stream) returns true, Bun.inspect(stream) prints state: 'readable', and finished(stream) never settles.
  • These consumers attach to the ByteStream or FileReader source in Rust and lock the stream with ReadableStream__lockNative (WebStreamsExports.cpp). The stream has no reader and no controller. The source ends the wire with self.sink.set(SinkHandle::None) only, so nothing changes m_state.

Fix

  • ReadableStream__lockNative stores the stream in a new lockedStream slot of the source wrapper. ByteStream::detach_sink and FileReader::detach_sink replace each sink.set(SinkHandle::None). They call NewSource::end_locked_stream before sink.end().
  • The stream closes when the source delivers its last chunk. It errors with the failure of the producer. It also closes when the sink stops early (abort, write failure, peer gone): pipeTo() cancels its source in that case.
  • The FileReader returns in wire_native_sink that end before the lock (Start::Err, Start::OwnedAndDone) now lock the stream too.
  • Verified: test/js/node/stream/web-stream-state.test.ts (9 new tests fail on main). Also the fetch, serve, spawn stdin, Bun.write, HTMLRewriter and S3 stream suites.

Background

  • A native stream is a JSReadableStream whose m_nativePtr is a wrapper cell around a Rust NewSource (ByteStream or FileReader).
  • A JS reader makes a controller and an adapter, which closes the stream. A native consumer skips both. It installs a SinkHandle on the source, which calls sink.write() and sink.end().
  • m_lockedWithoutReader makes .locked true without a reader. A consumed body keeps its lock, so the flag stays set.
Notes

Reproduction: the upstream server must still send the body when the consumer attaches. If the whole body is already buffered, ReadableStream::to_any_blob lifts it and no sink is wired. That path has the same symptom and is fixed separately.

import { isReadable } from "node:stream";
import { finished } from "node:stream/promises";
const res = await fetch(slowUrl);
const body = res.body;
await Bun.write("/tmp/out", res);
isReadable(body); // true before, false after
await finished(body); // never settles before

Fail before: with src/ from main (b99371011f) and bun bd, the 9 new tests fail and the 2 existing tests pass. With this branch, all 11 pass.

State after each end of the wire:

eventstate
source delivers the last chunkclosed
producer fails (on_data(Err), on_reader_error, stored read error, open failure)errored with that error
sink reports Done or a write errorclosed
sink closes early (cancel_from_sink, SourceHandle::close)closed
stream is cancelled from Rust (ReadableStream::done())closed
sink detaches (unpipe_without_deref)closed

The error that the stream stores is the same object that the consumer rejects with for a ByteStream (StreamError::JSValue).

Bun__NativeStreamSource__endLockedStream calls readableStreamCloseIfPossible and webStreamControllerError. Both are no-ops for a stream that is not readable, so ReadableStream::error() and cancel_with_reason() followed by done() keep the first state.

The S3 upload in s3/client.rs and the Bun.serve() response in RequestContext.rs wire the sink without wire_native_sink. Both call lock_native, so both get the slot. The inline ends in s3/client.rs now go through detach_sink.

The new GC edge (wrapper to stream) exists only while a sink is wired. end_locked_stream clears it.

Why a new slot: the fix needs an edge from the source to its stream, and the only edge today goes the other way (m_nativePtr). closeAdapter cannot be shared: NewSource::on_close clears it and passes the value to a microtask that does uncheckedDowncast<JSNativeStreamSourceAdapter>, and FileReader::on_reader_done calls on_close while a back-pressured sink is still attached. pendingPromise roots the pull promise. owner and sinkOwner root the HTMLRewriterTransform cell. The slot exists on the Bytes and File wrappers only. A Blob source never gets a native sink.

Suites run on the debug build: test/js/web/fetch/{body,body-stream,body-clone,fetch.stream,fetch-backpressure,fetch-abort-stream-body,fetch-stream-cancel-leak,blob-write,fetch}.test.ts, test/js/web/streams/{streams,native-source-onclose-leak,readable-stream-terminal-barrier-release}, test/js/bun/http/{serve,bun-server,serve-stream-body-error,serve-body-leak}, test/js/bun/spawn/spawn-stdin-readable-stream.test.ts, test/js/bun/io/bun-write.test.js, test/js/workerd/html-rewriter.test.js, test/js/bun/s3/{s3-upload-stream-gc,s3-stream-cancel-leak,s3-stream-error-gc,s3-write-to-file-sync-close,s3-connection-close}. The new tests also pass with BUN_JSC_validateExceptionChecks=1. The S3 upload consumer was checked by hand against a local mock (closed after the fix, readable before).