More TS-backed Streams Work - #7071
Open
jasnell wants to merge 16 commits into
Open
Conversation
Buffer-backed streams (string, ArrayBuffer/view, Blob, and URLSearchParams bodies) previously always constructed the legacy C++ ReadableStream, even under the typescript_implemented_streams compat flag, mixing the two implementations. Stream construction over a native source now goes through a helper shared by create() and bufferBackedImpl() that wraps the source in a ReadableStreamNativeSource and constructs the TypeScript stream when the flag is enabled. Covered by new C++ tests (TS-backed brand check via a fail-loudly wrapper stand-in, consumption, tryClone rewind, tee buffer carry, pumpTo extraction) and JS integration tests (Response/Request body instanceof + round-trips, URLSearchParams and empty bodies).
…s flag The Iterable/AsyncIterable BodyInit extension (fetch_iterable_type_support) previously called ReadableStream::from() directly, minting a legacy stream even under the typescript_implemented_streams compat flag. Body::extractBody now goes through the new JsReadableStream::from(), which dispatches on the flag: the legacy arm delegates to ReadableStream::from() unchanged, while the TypeScript arm constructs a TS stream over a C++-built JS underlying source driving the generator with the same algorithm (demand-driven highWaterMark-0 pulls, one generator.next() per pull, promise-typed values awaited before enqueue, close on completion, cancel forwarded to the generator's return()). The generator arrives pre-consumed from the OneOf unwrap (the iterable's iterator method has already been invoked), so the TS arm drives the captured jsg::AsyncGenerator rather than re-consuming the original object. The ts-webstreams wd-test now enables fetch_iterable_type_support explicitly so all variants (including the oldest-compat-date default) exercise the iterable arm. Covered by JS integration tests: async generator / sync iterable / promise-value bodies, reader-driven reads, cancel-to-return() forwarding, and error propagation.
detach() takes over a stream's internal state into a fresh stream, leaving the original a permanently locked, disturbed husk. It backs new Request(req) body proxying and the socket takeover paths (startTls/takeConnectionStream), which previously hit KJ_UNIMPLEMENTED for TypeScript-backed streams. The new detachReadableStream cppExport performs the takeover atomically with legacy-exact precondition texts, dispatching on the stream's state and backend: * Closed/errored: a state-copy shell (tee's non-readable precedent); the underlying source is not touched. * Native-backed: extract the source and reconstruct via ordinary native construction. A fresh conduit is required -- the old conduit's stream hooks close over the original stream -- and matches the legacy internal controller's detach, which builds a fresh controller over the removed source. expectedLength is re-read live, so residual accounting (including stashed bytes) stays exact. * Queued: tee's transfer recipe as a move -- a private-symbol shell adopts the shared controller and a cursor at the original cursor's exact position (added before removeCursor to avoid the all-cursors-gone hook). Tee-branch relationships (#onCancel, #onBranchSettled) move to the detached stream and are cleared on the husk, so composite cancels keep working after a branch is detached and husk transitions cannot fire the tee wiring. The C++ arm carries the retransmit buffer forward like the legacy arm, so detached buffer-backed streams remain rewindable. Known-unreachable edge, documented in the plan: an IgnoreDisturbed detach with an abandoned pull still in flight followed by an immediate read would trip the C++ source's defensive pullInFlight guard (sockets discards the detach result; http never passes IgnoreDisturbed). Covered by C++ tests (native/buffer/queued takeover, partially-read cursor transfer, disturbed/locked precondition throws, closed-stream state copy, tee-branch composite-cancel carry) and JS tests (new Request(request) body proxying for string and stream bodies).
The TypeScript arm of JsReadableStream::cancel() routed straight to the lock-blind internal cancel, so canceling a locked stream succeeded where the legacy arm rejects (matching ReadableStream.prototype.cancel). The arm now composes the locked check (legacy-exact rejection text) with the internal cancel from cppExports operations -- atomic under the isolate lock, and immune to user patching of the prototype method. forceCancel() keeps the lock-blind call: forcible teardown is its documented contract. Covered by C++ tests: locked cancel rejects then forceCancel succeeds and disturbs; unlocked cancel resolves and disturbs (both previously untested on the TS arm).
The TypeScript arm of JsReadableStream::tryGetLength() ignored the encoding parameter and always reported the controller-level (identity-byte) expected length. For a non-identity query -- Response::send() asks with the negotiated content encoding -- that reported an identity byte count as an encoded length, i.e. a wrong Content-Length for pass-through encoded bodies. The arm now dispatches on the encoding: * IDENTITY: the controller-level expected length, as before (both backends). * Non-identity: forwarded to the native underlying source (legacy internal-controller parity) through a new non-detaching source accessor (conduit peekSource -> nativeControllerPeekSource -> getReadableStreamNativeSource cppExport) and a C++-only ReadableStreamNativeSource::tryGetLength(encoding), which answers only while no identity bytes are stashed. Queued streams answer kj::none: an identity byte count is never a valid encoded length. (The legacy JS controller ignores the encoding and reports its identity expectedLength anyway -- a wire-protocol footgun deliberately not reproduced.) getExpectedLength() now delegates to the new method for the identity case. Covered by C++ tests: a source reporting distinct identity/gzip lengths answers each query correctly; queued byte streams with a declared expectedLength and buffer-backed streams answer none for encoded queries.
The onEof TypeScript arm returned a promise that never resolved, so a socket's allowHalfOpen:false EOF detection would silently hang under the typescript_implemented_streams flag. The signal now fires at the native conduit's source-driven close choke point (the closeStream hook, called only from the fused close-commit and closeFromSource paths): any observation of the native source's EOF through the conduit resolves the subscription -- reader reads, async iteration, and DrainingReader consumption alike. Cancel and error take other paths and never fire it; extraction-based pumps (pumpTo, the native+native pipe fast path) detach the source before its EOF could be observed; queued streams never fire it, matching the legacy JS controller, which has no EOF signal. Arming after the stream already closed never resolves, and detach() leaves the subscription with the husk -- both matching the legacy signalEof/eofResolverPair behavior. One deliberate divergence (decision log D7): the signal fires when C++ consumption helpers drain a native stream to EOF. The legacy readAllBytes detaches the source first and stays silent there; the TypeScript helpers read through the conduit, and observing a remote EOF during consumption serves the allowHalfOpen:false contract. Covered by C++ tests for all five cells: reader-driven fires, consumption fires, cancel does not, extraction pump does not, queued never does.
The #pendingClosure flags were write-only signals: a Socket entering close()
marked both of its TypeScript-backed streams, but nothing consulted the
marks, so reads and writes kept racing the teardown where the legacy
internal controller fails them fast.
The TypeScript streams now implement the same gates with the legacy error
texts ("This {Readable,Writable}Stream belongs to an object that is
closing."):
* Readable: reader reads (default and BYOB), draining reads (covering the
C++ consumption helpers and the queued pump), pipeTo (and thereby
pipeThrough), and tee.
* Writable: write() only, checked before the size algorithm runs so no user
code executes for a write against a closing socket.
Deliberately not gated, matching legacy: getReader/getWriter acquisition,
cancel, abort, and the teardown's own forceFlush/forceClose sequence, which
runs after the flag is set. The gates apply uniformly to both backends at
the stream layer; only sockets (native-backed) ever set the flag.
Covered by C++ tests: default and BYOB reads reject; consumption rejects
and tee throws; pipeTo rejects while cancel still resolves; writable write
rejects while forceFlush/forceClose still complete and end the sink.
Red-green verified by neutering the setters.
…the output gate getUnderlyingForTest() returned the raw legacy WritableStream and was KJ_UNIMPLEMENTED for TypeScript-backed streams, so socket tests could not run under the typescript_implemented_streams flag. Its sole consumer (sockets-test's output-gate test) now uses the backend-neutral writeForTest(js, chunk): the legacy arm writes through the controller as before; the TypeScript arm acquires the writer, writes, and releases it. Running the output-gate test against the TypeScript backend immediately exposed a real bug: WritableStreamNativeSink performed its sink I/O without consulting the Durable Object output gate, so a flagged actor's socket writes could become externally observable while an output lock was pending -- exactly what the gate exists to prevent. The legacy internal controller stores an output lock with every queued write/close/flush/pipe event and awaits it before touching the sink. WritableStreamNativeSink now waits for pending output locks at each of its I/O points, mirroring the legacy events: * write(): before the sink write (covers writer writes and every pipe that drives the TS writable machinery). The sink reference is safe across the wait: writeInFlight defers abort()/detach() release to settlement. * closeImpl(): before end(). * pipeFrom(): once, before starting the native+native pump -- the pump coroutine is created inside the lock continuation, since KJ coroutines run eagerly and a pre-created pump would progress regardless of chaining. The flush queue marker remains ungated (its writes are themselves gated; an empty-queue flush resolving early is a documented timing delta from the legacy Flush event). Coverage: the output-gate socket test now runs against both backends (runSocketWriteOutputGateTest), and new actor-fixture tests pin each gate independently -- write, close-only (not vicariously blocked behind a gated write), and the pipeFrom fast path. writeForTest itself is covered on both backends. All gates red-green verified.
Adds coverage the bridge-test audit flagged as missing: * Readable: the arrayBuffer/bytes/json/blob consumption helpers against TypeScript-backed streams; addRef sharing; pipeThrough with a hand-assembled pair of TypeScript endpoints. * WritableStreamNativeSink: abort during an in-flight write (the deferred pendingAbort release, driven at the hook level -- the TS machinery serializes write/abort, so the overlap is defensive-only); SharedArrayBuffer chunks; the byte-based sizing strategy (highWaterMark-configured create reports byte-accurate desiredSize). * Pipes: a mid-pump signal abort of the native+native fast path (only the pre-aborted case was pinned), verifying rejection plus the option-gated source cancel. * JsReadableWritablePair::jsgTryUnwrap tier-2 (dictionary-shaped) unwrap with TypeScript members via a wrapper stand-in whose member unwraps delegate to the TS brand checks, plus the either-member-fails and non-object rejections. (The pair has no production consumer yet; this pins the fallback for the coming TransformStream-acceptor migrations.) Buffer-backed constructor coverage for Blob/BufferSource/URLSearchParams bodies already landed with the P1 wd-tests, which exercise those constructors end to end through Body::extractBody.
The iterator-protocol object checks in from() used isActualObject, which excludes callables, but the spec's "Type(x) is Object" includes them: GetIterator accepts an iterator method returning a function that carries next(), and IteratorNext/return results are validated the same way. The checks now use an isObjectLike helper that admits callables, fixing the two WPT from.any.js function-iterator cases (previously expected failures); the option-bag validations deliberately keep isActualObject. The two remaining piping expected failures (abort-after-close ordering and flow-control backpressure cadence) share one root cause, now documented in the WPT config: the pump drains in batches and observes source-close through its read loop rather than through closedPromise reactions. Fixing them requires reshaping the pump around chained per-chunk reads -- spot patches to the reaction or the drain size break other timing-sensitive piping suites.
Cleanup items: * streams.ts no longer re-exports kNativeSource / kExtractNativeSource / kNativeSink / kExtractNativeSink. They existed so JS mocks could build native-marked sources before the C++ handshake landed; the C++ tests now construct real ReadableStreamNativeSource objects and the bridge reaches the extraction symbols through the API-symbol registry. The ReadableStreamDrainingReader export stays: main.ts installs it under the internal-testing expose_draining_reader flag. Stale TEMPORARY notes in native.ts and both AGENTS.md files updated to describe the current state (including per_isolate/AGENTS.md's outdated claim that main.ts installs a dev-only lazy globalThis.streams surface). * PrefixedSource now implements pumpTo(): the kj-heap prefix is written in the pre-proxy phase and the inner source's pump is delegated to with its deferred-proxy classification passing through (the cache.c++ delegation pattern), so the rare stash-prefix pump path no longer loses deferred proxying by falling back to the generic pump loop. * The two piping WPT expectations are reworded as INTENTIONAL SPEC DIVERGENCE: the batched-drain pipe pump deliberately amortizes per-chunk read overhead (one drained batch per writer-ready cycle) rather than following the reference implementation's one-chunk-per-read shape; the flow-control desiredSize cadence and the abort-after-close ordering are consequences of that design, not defects. The getCppExport per-call lookup was assessed and left as-is: it is a cached-module map find plus one property get; a per-context handle cache is not worth new machinery without profiling evidence.
The writable-side marker comment described pipeFrom as the sink's one extension beyond the standard hooks, but WritableStreamNativeSink also implements detach() (consumed by detachWritableStream when the underlying connection is taken over). List both.
The deferred-proxy contract (api/deferred-proxy.h) promises that when a pump's proxy is not a no-op, the flow can continue without pinning the isolate or the IoContext -- the property io/worker-entrypoint.c++ Stage 3 relies on when it awaits the proxy task after dropping the incoming request. No test proved it: deferred-proxy-test.c++ covers coroutine machinery only, every pump test flattens outer+proxyTask inside a live IoContext, and every existing pump test uses plain heap sinks that take the no-op path, so no test's deferred phase has ever carried data. The new deferred-proxy-flow-test drives the real (system-to-system) proxy through both backends using a TestFixture affordance: runInIoContext unwraps one promise layer, so a callback returning kj::Promise<DeferredProxy<void>> hands the DeferredProxy back by value after the IncomingRequest -- the IoContext's sole owner -- is destroyed, reproducing the production Stage-3 shape. A gated test-owned source guarantees the flow cannot complete in-request: * Receiving the DeferredProxy with the gate still closed is itself the shape proof (a no-op's outer promise resolves only at EOF, so it would deadlock on the gate -- verified red: degrading tryPumpFrom to kj::none turns both real-proxy tests into timeouts). * The pre-gate bytes flow, the gate releases, and the rest of the data flows to the far end of a test pipe entirely after IoContext destruction; any lingering IoOwn tether would trip its far-get check. * Environment probes inside the source's reads verify every deferred-phase operation runs with the isolate un-entered and no current IoContext. (Pre-split probes are exempt and documented: KJ coroutines start eagerly, so the pump's first read typically executes synchronously inside pumpTo() under the Worker lock.) The TypeScript variant additionally proves the extraction plumbing (kExtractNativeSource -> releaseForPump -> pumpExtractedSource) fully de-tethers the source from the IoContext. Companion no-op tests pin the opposite shape for buffer-backed (legacy) and queued (TS) pumps, so a change that silently flips a pump between real and no-op fails loudly.
A verification pass on the new proof tests found two overclaims and one broken discriminator: * The real-proxy test claimed the pre-gate bytes flow after IoContext destruction. Not provable: the pump starts eagerly inside pumpTo(), so its first read (and the rendezvous delivery) may complete in-request. The comments now scope the post-destruction guarantees to everything from the gate onward, which is what the probes actually anchor. * Only source reads were probed; sink-side operations were asserted by proximity. A ProbingOutputStream now wraps the test pipe's write end, recording the environment at every write and at sink teardown -- and, by hiding the pipe's pump adoption, forces the kj pump into its plain read/write loop so both sides are deterministically observed. The deferred-phase probe minimum rises to four (post-gate read, post-gate write, EOF read, sink teardown). * The no-op pins did not discriminate: polling the proxy task to completion after the fact succeeds for a REAL proxy over small in-memory data too. They now assert the defining property in-request -- with the far-end reader withheld, the flow parks and the OUTER promise must remain pending, which an eagerly-resolved real outer fails loudly. The first version of that check was itself vacuous (a plain .then() chain is lazy and never ran its continuation), caught by mutation testing and fixed with eagerlyEvaluate. Verified red both ways: resolving the queued pump's outer eagerly fails exactly the pending check; degrading tryPumpFrom to kj::none times out the real-proxy tests. The pass also confirmed at the code level why post-destruction flow is legitimate: EncodedAsyncOutputStream::tryPumpFrom builds the entire deferred segment from the RAW inner streams; the wrapper methods that register pending events on the IoContext are not part of it. This is now recorded in the test. Remaining limitation, stated for honesty: non-ASAN runs cannot prove the absence of benign dangling-IoContext reads in the deferred phase; code inspection shows none, and the asan test variants are the memory-safety net.
![ask-bonk[bot]](https://avatars.githubusercontent.com/in/2454580?s=60&v=4){kind=small}
Contributor
|
I'm Bonk, and I've done a quick review of your PR. PR #7071 expands TypeScript-backed streams and C++ bridge coverage.
|
The pending-closure gate lived only in readableStreamPipeTo's precondition block, but ReadableStream.prototype.pipeThrough reaches the pipe machinery through readableStreamPipeThroughTo -> pipeToInternal, bypassing it. An ungated pipeThrough on a closing socket's readable locked both endpoints and let the read-side gate reject inside the pump, whose error shutdown then aborted the transform's writable -- where the legacy controller's gate (checked in its pipeTo, which both legacy entry points funnel through) rejects before any locking and leaves the transform untouched. readableStreamPipeThroughTo now applies the gate before pipeToInternal. It rejects rather than throws, matching the legacy surface exactly: the prototype marks the pipe promise handled, so the result is an untouched transform and a hidden rejection. The gate in readableStreamPipeTo stays: it also guards the native+native fast path, which returns before reaching this junction; the pipeTo route re-checking here is a harmless boolean read. The regression test drives the JS prototype pipeThrough directly (the C++ bridge's pipeThrough composes over the already-gated readableStreamPipeTo, which is why existing tests missed this). Lock checks cannot discriminate -- the ungated pump's finalize releases both locks after the read gate rejects -- so the decisive probe is the transform's continued usability: a write to the pair's writable must still succeed afterward, which fails with the pending-closure error on the ungated path (red-verified).
Collaborator
Author
|
/bonk update your review |
Contributor
|
LGTM! |
readability-container-size-empty: use !stash.empty() rather than stash.size() > 0 in ReadableStreamNativeSource::tryGetLength. google-readability-casting: use static_cast<uint64_t> for the highWaterMark argument in the byte-sizing test.
jasnell
commented
Aug 19, 2026
| // support deferred proxying) and retains the buffer for retransmission. The stream over the | ||
| // source is constructed through the same compatibility-flag dispatch as create(), so under the | ||
| // typescript_implemented_streams flag buffer-backed streams are TypeScript-backed like every | ||
| // other stream. |
Collaborator
Author
There was a problem hiding this comment.
Note: these "buffer-backed" streams are what is created when we have constructions like new Response('foo'). The actual data buffer originates in the JS heap. In this path currently, we end up passing that out to a ReadableStreamSource then back into the ReadableStream via a native underlying source. that's wasteful. As a follow on step, when TS streams are used, we can skip the native underlying source and go directly to a fully TS/JS stream since deferred proxy is not available under this case anyway. Will be faster. This will be a follow on optimization tho.
jasnell
commented
Aug 19, 2026
| // arm's closures, these capture it without GC visitation: the captured references are | ||
| // strong, keeping the (JS-unreachable) generator state alive as long as the stream's | ||
| // source object is. | ||
| auto rcGenerator = kj::rc<jsg::AsyncGenerator<jsg::Value>>(kj::mv(generator)); |
Collaborator
Author
There was a problem hiding this comment.
There's an inefficiency here. We grab the async iterator from JS, wrap it in C++ classes, then wrap that in C++ functions just to drive the TS implemented stream. We can shortcut this by adding a callout to TS. That will require a bit of refactoring of the jsg::AsyncGenerator design however since it extracts the iterator eagerly. This will be a follow up optimization.
guybedford
approved these changes
Aug 20, 2026
guybedford
left a comment
guybedford
left a comment
Contributor
There was a problem hiding this comment.
Findings (none blocking):
- The native+native pipe fast path doesn't exclude a destination with a close queued/in flight, diverging from legacy
tryPipeFromand opening a sink lifetime window (inline oncloseImpl). - Comment placement nit in readable.ts (inline).
- Question on prototype dispatch in
from()'s pull (inline).
Note: this review was performed with AI assistance under my direction; details in the inline comments.
Comment on lines
+555
to
+561
| // Durable Object output gate: like the legacy controller's queued Close event, the | ||
| // sink must not be ended while an output lock is pending. The sink reference stays | ||
| // valid across the wait: the TS machinery serializes sink operations, and abort() | ||
| // only releases the sink itself. | ||
| kj::Promise<void> endPromise = nullptr; | ||
| KJ_IF_SOME(lock, ioContext.waitForOutputLocksIfNecessary()) { | ||
| endPromise = lock.then([&sink = *active.sink]() { return sink.end(); }); |
Contributor
There was a problem hiding this comment.
The "TS machinery serializes sink operations" justification doesn't cover pipeFrom extraction, which bypasses the sink-hook serialization. Legacy WritableStreamInternalController::tryPipeFrom rejects when the destination isClosedOrClosing() ("This destination writable stream is closed."), per the spec's closing-must-be-propagated-backward step. The TS fast-path dispatch gate only checks getState(destination) === 'writable', which stays 'writable' while a close is queued/in flight, and extractNativeSink checks only lock + marker. So:
const w = sock.writable.getWriter();
w.close(); w.releaseLock(); // close hook dispatches; end() in flight (or parked on this gate)
body.pipeTo(sock.writable); // fast path extracts; pipeFrom moves the sink Own outTwo consequences:
- Behavioral: the pipe pumps instead of rejecting (legacy/spec reject).
- Lifetime: the pending
end()— including this gate-parked[&sink = *active.sink]continuation — references a sink whose ownership moved to the pump; when the pump completes and destroys it with theend()work still pending, that's a use-after-free.writeInFlightprotects the write path (pipeFromasserts it), but nothing tracks close-in-flight.
Suggest gating the TS pipe dispatch's fast path on !isWritableStreamClosedOrClosing(destination) (falling back to the JS pump, which propagates closing backward correctly per the passing WPT), and/or adding close-in-flight tracking to pipeFrom's preconditions alongside the !writeInFlight check. Fine as a follow-up given the flag is experimental.
Comment on lines
+689
to
+696
| // The user-visible error for reads/pipes/tees attempted after the stream's | ||
| // owning object initiated closure (see #pendingClosure). The text matches | ||
| // the legacy internal controller's exactly. | ||
| function pendingClosureError(): TypeError { | ||
| return new TypeError( | ||
| 'This ReadableStream belongs to an object that is closing.' | ||
| ); | ||
| } |
Contributor
There was a problem hiding this comment.
Nit: pendingClosureError() landed between defaultReaderReadInternal's doc comment (the "default-read core" + "BACKEND-BLIND" block above) and the function it documents, so that block now reads as documentation for pendingClosureError. Suggest moving this function above the whole comment block.
| return js.v8Ref<v8::Value>(js.v8Undefined()); | ||
| }); | ||
| } | ||
| webstreams::invokeMethod(js, controller.getHandle(js), "enqueue"_kj, jsg::JsValue(handle)); |
Contributor
There was a problem hiding this comment.
Question: invokeMethod(controller, "enqueue"/"close") does a prototype-chain get on a controller whose prototype is a user-reachable global under the flag, so replacing ReadableStreamDefaultController.prototype.enqueue turns iterable-body streams into "method not found" internal errors. It's consistent with the TS-internal from() (which also dispatches controller.enqueue(...)), but it contrasts with the captured-call discipline this PR applies to cancel(), and the invokeMethod header comment says receivers are "module-owned TypeScript code, not user objects". Is prototype dispatch here intentional?
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
2 participants
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Multiple sets of changes here, all divided into individual commits for review. Each commit has it's own detailed description.
Draft for now as I'm still reviewing the full set of changes myself.agent output fully reviewed by me... ready for others to review!