CompressionStream/DecompressionStream: emit each chunk's output in bounded steps - #38695
Conversation
…unded steps The native coder drained the whole expansion of an input chunk into one buffer before the consumer's first read() resolved, so a few hundred bytes of brotli or zstd (or 4 KiB of gzip) expanding to gigabytes were materialized at once; TransformStream backpressure only applies between input chunks. A streaming size guard in the consumer could not pre-empt it. The coder now transforms a chunk (or the flush) in steps of at most 64 KiB of output (1 MiB on the thread-pool path, or the chunk's own size if larger) and keeps the unconsumed tail of the chunk itself. The C++ arm delivers each step's output and, when the readable side or the native sink is full, parks the chunk on the readable's backpressureChangePromise or the sink's ready promise and resumes from there; the chunk's transform promise (and so its write()) settles once the whole expansion has been consumed. While a chunk is pending across turns, ClearAlgorithms defers the coder release to the chunk's terminal, which also covers the close algorithm clearing algorithms while a large flush is still parked.
|
Warning Review limit reachedYou’ve reached a temporary PR review limit under our Fair Usage Limits Policy. Next review available in: 29 minutes Enable usage-based reviews in Billing to review now. Otherwise, wait until the next included review is available. How can I continue?After more reviews become available, a review can be triggered using the To avoid repeated limits, reduce automatic review volume by pausing incremental auto-reviews earlier, using label-based review opt-in, excluding WIP or generated PR titles, or requesting reviews manually when the PR is ready. If your team needs uninterrupted high-volume reviews, an organization admin can enable usage-based reviews. How do review limits work?CodeRabbit enforces per-developer PR review limits for each organization. Most developers receive the normal plan review availability. For paid Pro and Pro+ PR reviews, CodeRabbit uses adaptive limits for sustained high-volume activity. When a developer's recent PR review activity reaches the 95th percentile or higher among CodeRabbit users, additional reviews become available more gradually as earlier reviews age out of the rolling window. Please refer docs for additional details. Review details⚙️ Run configurationConfiguration used: Path: .coderabbit.yaml Review profile: ASSERTIVE Plan: Pro Run ID: 📒 Files selected for processing (2)
No actionable comments were generated in the recent review. 🎉 ℹ️ Recent review info⚙️ Run configurationConfiguration used: Path: .coderabbit.yaml Review profile: ASSERTIVE Plan: Pro Run ID: 📒 Files selected for processing (13)
WalkthroughCompression and decompression streams now process output in bounded codec steps. The implementation supports continuation, backpressure, synchronous and asynchronous delivery, cancellation, native sinks, deferred cleanup, and configurable ChangesCompression stream stepping
Possibly related PRs
Suggested reviewers: 🚥 Pre-merge checks | ✅ 4✅ Passed checks (4 passed)
Comment |
|
Status: ready for review (head 484a9cd, an empty retrigger on top of eb0238d). Reproduced on bun 1.4.0 with a 4 MiB-to-5 GiB brotli body decoded through Tests: the CI: in build 97585 every lane that has run passed (176 jobs, including the Linux ASAN lanes and both Windows lanes for Revisions: 57eb268 fixed the cancel-after-close hang found in review; 330d444 fixed two exception-scope validation failures caught by CI; 734ac8c is the rework requested in review (per-stream |
![claude[bot]](https://avatars.githubusercontent.com/in/1236702?s=60&v=4){kind=small}
transformStreamDefaultSourceCancelAlgorithm returns the in-flight close's finish promise without running the cancel reaction, so its unblock-write step never resolved the backpressure promise a multi-step flush was parked on, and close() and cancel() both stayed pending. Perform the unblock in that branch too; nothing else can be waiting on the promise once a close has started.
![github-actions[bot]](https://avatars.githubusercontent.com/in/15368?s=60&v=4){kind=small}
|
Pushed as 734ac8c; description rewritten for the new shape.
|
writer.abort() (or an aborted pipeTo) waits for the in-flight write, and a chunk still being drained into an unread readable would never finish it. Abandon it from writableStreamStartErroring when a write is in flight; an in-flight close (a flush) is left to finish, as a close in progress wins over the abort. This also covers the transform error terminal, whose own abandon call goes away.
…ring as well writableStreamDefaultControllerError cleared the sink algorithms before calling writableStreamStartErroring, so the check there never saw a transform sink on that route (node:stream's addAbortSignal() on the writable reaches it) and a chunk parked mid-expansion kept the in-flight write pending forever. Clear the algorithms after StartErroring instead; nothing else in it reads them while the stream is still Writable.
There was a problem hiding this comment.
I reviewed the current head (eb0238d) and didn't find any new issues — the two hangs flagged earlier (writer.abort() and the addAbortSignal() route through writableStreamDefaultControllerError) are both fixed with tests, and the userJS annotations are corrected. Given the scope (a new cross-turn stepping protocol wired into the shared TransformStream/WritableStream spec algorithms, a spec-order swap in writableStreamDefaultControllerError, coder-lifecycle deferral across the thread pool, and a new user-facing constructor option), a human sign-off is still warranted.
Checked this pass: the reordered ClearAlgorithms in writableStreamDefaultControllerError against its other callers and against writableStreamFinishErroring reachability; that every route into a parked chunk (pull, sink onReady, StartErroring, source cancel, sink detach) now reaches Continue or Abandon; and that nativeTransformReleaseStateIfIdle gates on all four busy conditions so takeCodecPromise cannot free the coder under an in-flight pool step or a stacked arm.
Extended reasoning...
Overview
This PR restructures CompressionStream/DecompressionStream so one input chunk's output is emitted in bounded steps rather than materialized whole, fixing a decompression-bomb OOM. It touches 15 files: the Rust coder (CompressionStreamCoder.rs, transform → step with a Pending tail), the C++ transform arm (JSCompressionStreamShared.cpp, rewritten around a CodecOutcome state machine and m_codecPromise), and hooks into the shared streams machinery — transformStreamDefaultSourcePullAlgorithm, transformStreamDefaultSourceCancelAlgorithm, writableStreamStartErroring, writableStreamDefaultControllerError, the native-sink onReady/detach in BunStreamSource.cpp, and the coder-release deferral in JSTransformStreamDefaultController.cpp. It also adds a Bun-extension constructor option ({ highWaterMark }) with type declarations, and 28 new tests.
Security risks
The motivating bug is itself a DoS vector (a few-hundred-byte request body OOM-killing a server), and the fix mitigates it. The new highWaterMark option is validated through extractHighWaterMark (RangeError on NaN/negative, TypeError on non-object) and floored at 1 in Rust; Infinity restores the old unbounded behavior explicitly. No new untrusted-input parsing.
Level of scrutiny
High. This is not a mechanical change: it introduces cross-turn native state (m_codecPromise) whose lifecycle every terminal of both the readable and writable sides must now settle, deliberately reorders two calls in a spec algorithm (writableStreamDefaultControllerError now runs StartErroring before ClearAlgorithms), and threads a new deferral condition through the eager-release path shared with TextEncoderStream/TextDecoderStream. Two blocking hangs were found and fixed during review already. The maintainer has been actively shaping the design (configurable HWM, removing the extra promise), so a human should confirm the final shape matches what was asked for.
Other factors
All prior review threads are resolved. Test coverage for the new behavior is thorough (per-format stepping, all three abandon routes, the multi-step-flush-vs-abort ordering, the thread-pool path and its error arm, native-sink delivery, option validation, client-disconnect propagation). The PR description states the WPT streams suite, streams.test.js, TextEncoder/DecoderStream tests, and node webstreams compression tests pass under BUN_JSC_validateExceptionChecks=1. The writableStreamDefaultControllerError reordering is justified in the code comment and the author's reply (a Writable-state stream has no pending abort request, so StartErroring on this path cannot reach FinishErroring's [[AbortSteps]]), but it is a spec deviation in shared code and worth a maintainer's eye.
| return promise; | ||
| } | ||
| } | ||
| RELEASE_ASSERT_NOT_REACHED(); | ||
| } | ||
|
|
||
| void nativeCodecContinue(JSGlobalObject* globalObject, JSTransformStream* stream) | ||
| { |
There was a problem hiding this comment.
🟡 Third variant of the same abort-path deadlock (siblings fixed in 65c5929 and 9214ac8): if writer.abort() runs re-entrantly from inside the initial stepChunkHere loop — reachable only via an adversarial Object.prototype.then getter fired by the first step's enqueue resolving a pending reader.read() — writableStreamStartErroring's nativeCodecAbandon hook is a no-op (m_codecPromise not set yet), stepChunkHere's own abandon check doesn't fire (continuation==false), and transformChunk then parks m_codecPromise with the writable already Erroring, so writer.abort(), writer.closed, and the coder release stay pending forever unless the user reads again. Not reachable from ordinary code, but it's the same class REVIEW.md asks to close in one PR; a check of stream->m_writable->m_state (or setting m_codecPromise before entering stepChunkHere so the abandon has something to clear) in the Pending case closes it.
Extended reasoning...
The gap the two earlier fixes leave
Commits 65c5929 and 9214ac8 hook nativeCodecAbandon into writableStreamStartErroring (WritableStreamOperations.cpp:314-318) so that a writer.abort() / addAbortSignal() while a chunk is parked mid-expansion gives the chunk up. That hook is keyed on stream->m_codecPromise being set:
void nativeCodecAbandon(JSGlobalObject* globalObject, JSTransformStream* stream)
{
if (!stream->m_codecPromise)
return;
settleCodecChunk(globalObject, stream, JSValue());
}But transformChunk's on-thread arm only sets m_codecPromise after stepChunkHere returns Pending (JSCompressionStreamShared.cpp:293-296). So if writableStreamStartErroring is reached re-entrantly from inside the initial stepChunkHere loop, the abandon is a no-op, and transformChunk then creates m_codecPromise with the writable already in state Erroring — a state StartErroring will never re-enter (it asserts state == Writable).
stepChunkHere itself has an abandon-detection check, if (continuation && !stream->m_codecPromise) return CodecOutcome::Pending;, but continuation was captured as !!stream->m_codecPromise at entry — false on the initial call — so it does not apply.
Step-by-step trace
- Setup.
ds = new DecompressionStream("brotli");writer = ds.writable.getWriter();reader = ds.readable.getReader(). InstallObject.defineProperty(Object.prototype, "then", { get() { writer.abort(reason); return undefined; } }). Callreader.read()(queue empty → read request added to[[readRequests]], pull algorithm flips[[backpressure]]false), thenwriter.write(bomb). - Entry. A microtask runs
onTSSinkWriteBackpressureChangeFulfilled→PerformTransform→runNativeArm(setsm_nativeStateInUse=true) →transformChunk→stepChunkHerewithcontinuation = !!m_codecPromise = false.ProcessWritehas already setm_inFlightWriteRequest. - First step runs the getter.
runStepHere→transformStreamDefaultControllerEnqueue→readableStreamDefaultControllerEnqueuesees the pending read request →readableStreamFulfillReadRequest→JSReadRequest::chunkSteps(kind::Promise) →resolvePromise(readPromise, {value,done}). Per this header's own note onresolvePromise, resolving with any object performsGet(v, "then")— the getter runs synchronously and callswriter.abort(reason). - The abort's abandon is a no-op.
writableStreamAbort→ setsm_pendingAbortRequest→writableStreamStartErroring. The new hook fires (kind==Transform && m_inFlightWriteRequest), butnativeCodecAbandonseesm_codecPromise==nulland returns.StartErroringreturns withoutFinishErroring(write in flight). The readable is untouched. - The loop parks anyway. Back in
stepChunkHere:continuation==falseso the abandon-detection bailout is skipped;step.thrownis empty;consumerFull==false(the enqueue fulfilled a read, queue is still empty,m_backpressurestays false);step.more==true→ loop again. The secondrunStepHereenqueues into the now-empty queue (no pending read) →HasBackpressurebecomes true →transformStreamSetBackpressure(true).consumerFull==true→ returnsCodecOutcome::Pending. - transformChunk parks a promise the writable can never abandon.
case CodecOutcome::Pending:createsm_codecPromiseand returns it. The writable isErroringwithm_inFlightWriteRequestset;StartErroringwill never run again; the readable is stillReadablewith one chunk queued so no cancel; no sink attached. The only thing that can settlem_codecPromiseis anotherreader.read()driving the pull algorithm — if the user stops reading (they just aborted),writer.abort()'s promise,writer.closed, and the deferred coder release (nativeTransformReleaseStateIfIdlegated on!m_codecPromise) all stay pending forever.
Why this is a regression
Pre-PR codeAndEnqueue transformed the whole chunk in one call and always returned an already-settled promise (promiseFulfilledWith(undefined)), so a re-entrant abort during its single enqueue still let the in-flight write finish immediately and FinishErroring proceed.
Why nit, not normal
The only synchronous route from runStepHere's enqueue to user JS on the initial call is ReadRequestKind::Promise resolving with an iterator-result object → Get(v, "then") on Object.prototype (the header explicitly documents this; commit 172c237 flipped nativeCodecContinue/pullAlgorithm to userJS: yes for exactly this reason). PipeTo/Tee/AsyncIterator/ReadStreamIntoSink read-request kinds all defer via a microtask. So triggering this requires an adversarial Object.prototype.then getter — an attacker with that capability has already won, and the impact is a hang + leaked coder in the hostile process, not memory corruption or a security boundary. Not merge-blocking.
That said, REVIEW.md's "Fix the whole class in the same PR" applies (this is the third instance of the exact bug class this PR just iterated on twice), and "Anything that can run user JS can synchronously free your state — re-validate liveness guards after every callback" is precisely the rule the Pending case is missing.
Fix
Either:
- In
transformChunk'sPendingcase, checkstream->m_writable->m_state != WritableStreamState::Writable(the write can no longer usefully stay in flight) and returnpromiseFulfilledWith(undefined)instead of parking; or - Create and set
m_codecPromisebefore callingstepChunkHerein the on-thread arm (matching the off-thread arm), so a re-entrantnativeCodecAbandonhas something to clear andstepChunkHere's existingcontinuation && !m_codecPromisecheck catches it — thentransformChunkreturns the (now-cleared) promise fulfilled, or a fresh fulfilled one, whenstepChunkHerereportsPendingwithm_codecPromisealready gone.
The first is one line plus an include; the second reuses the existing detection but touches the Done/Failed/DoneSinkFull cases (they'd need to clear the pre-set promise).
There was a problem hiding this comment.
Confirmed on main (the probe delivers the first 64 KiB piece, abort() is called from the getter, and write()/abort() never settle). Since this PR has merged, the fix is in a follow-up: #38970. It takes your second option: transformChunk registers m_codecPromise before the first step, so the first call goes down the same path as a continuation (an abandon from inside a delivery resolves it and stepChunkHere stops), and the per-outcome switch in transformChunk is replaced by the existing settlePendingChunk. Test added with an Object.prototype.then getter; it times out on main and passes with the change.
The per-coder output Vec is replaced by one scratch Vec in RareData, taken for the duration of a JS-thread step and put back afterwards (kept up to 256 KiB), so N idle streams no longer each retain a highWaterMark-sized buffer. The thread-pool step owns its output Vec on the job ctx.
| /// One step of the chunk (or, with `finish`, the final flush) in progress: | ||
| /// collects at most `max(high_water_mark, chunk length)` bytes into `out` and | ||
| /// returns `true` if the codec stopped at that cap, in which case the caller | ||
| /// must step again (with no input) before feeding the next chunk. |
There was a problem hiding this comment.
If you need a paragraph-long comment to justify why the workaround is OK, the code is wrong — fix the code
| /// Drives the codec until the chunk is done or `out` holds `cap` bytes. | ||
| /// A `continuing` step (not the chunk's first) calls the codec even with no | ||
| /// input left, to drain the output it is holding. |
There was a problem hiding this comment.
If you need a paragraph-long comment to justify why the workaround is OK, the code is wrong — fix the code
2 participants
Problem
DecompressionStream(all five formats;CompressionStreamhas the same structure) materializes the entire expansion of one input chunk as one output chunk before the consumer's firstread()resolves. A 203-byte brotli chunk comes out of the firstread()as a single 268,435,456-byteUint8Array; a 4 KiB brotli request body expanding to 5 GiB OOM-kills a server whose handler enforces a 10 MiB limit in its read loop, because the limit is checked after the expansion exists.CompressionStreamCoder::transform(src/runtime/webcore/CompressionStreamCoder.rs) loops the codec until the chunk is fully consumed, collecting everything into one buffer, and the arm inJSCompressionStreamShared.cppenqueues it once. TransformStream backpressure only applies between input chunks, so nothing between the codec and the consumer can stop one chunk's expansion.DecompressionStreaminflates lazily perread(); the same guard there trips after the first 16 KiB.Fix
transformbecomesstep. A step collects at most the stream'shighWaterMarkbytes of output, or the chunk's own size if that is larger (the bound is on expansion; a big chunk still finishes in a step or two), and reports whether the codec stopped short. When it did, the coder copies the chunk's unconsumed tail into itself (Pending), so later steps take no input and do not depend on the JS buffer staying intact while user code runs in between. The per-format loops are unchanged apart from the cap check; the zstd split-magic join moves up intostepso the tail offset is uniform.new DecompressionStream("gzip", { highWaterMark })(alsoCompressionStream), read like a queuing strategy (sizeis ignored; NaN/negative is a RangeError, a non-object is a TypeError,Infinitydisables the splitting), default 64 KiB, the same default output size as the node:zlib stream adapter. It is stored in the coder at creation; there are no step-size constants left. Declared in bun-types.m_codecPromise(renamed fromm_asyncCodecPromise, which served the same purpose for the thread-pool path only): a pending transform promise is what keeps the write in flight, so the producer waits too. No other promise is involved: the consumer drives the rest through two functions,nativeCodecContinue, called from the readable's pull algorithm and from the native sink's onReady, andnativeCodecAbandon, called from whichever side goes away first: the writable starting to error while the chunk's write is in flight (writer.abort(), an abortedpipeTo, a transform error, node:stream'saddAbortSignal(); all routes go throughwritableStreamStartErroring, andwritableStreamDefaultControllerErrornow clears the sink algorithms after calling it rather than before so the check there still sees the transform; an in-flight close, that is a flush, is left to finish, since a close in progress wins over the abort), the readable's cancel algorithm (before its early return, so cancelling afterclose()with a flush in progress is covered), and a sink detach. Chunks that started on the thread pool keep stepping there;deliverAsyncre-dispatches while the consumer has room and drops the output of a step whose chunk was abandoned meanwhile.[[backpressure]], steps the pending chunk and returns null (pull complete) when the enqueues set backpressure again, otherwise the usual promise; either way the next pull still finds the state the spec algorithm expects. JS transforms take the existing path.transformStreamDefaultControllerClearAlgorithmsdefers the coder release whilem_codecPromiseis set (nativeTransformReleaseStateIfIdle), and the chunk's terminal releases it. The close algorithm clears algorithms as soon as the flush arm returns, so a brotli/zstd flush of more than one step depends on this.write()settles once its expansion has been consumed rather than once it has been produced (Node behaves the same for output beyond its buffer); a single write whose output fits in one step still completes without a reader, as before. Pieces are at mosthighWaterMark(or the chunk's size) bytes; readers that concatenate are unaffected.test/js/web/streams/compression.test.ts,bounded output per input chunkblock (28 tests: per-format piece size and pendingwrite(),reader.cancel(),writer.abort()andaddAbortSignal()mid-expansion,writer.abort()during a multi-step flush leaving the output intact, request-body guard, native response sink, thread-pool path and its error path, the option on both classes and its validation,Infinity, encoder side, multi-step flush into the sink,reader.cancel()afterwriter.close()in both the no-read and re-parked variants, and a client disconnecting mid-expansion propagating back to the pipeline's source); 21 of them fail on the unfixed build, the whole file (65 tests) passes with the fix; thewriter.abort()andaddAbortSignal()tests additionally hang with, respectively, thewritableStreamStartErroringhook removed and thewritableStreamDefaultControllerErrororder restored, which are the hangs they guard against. All of it, plus the vendored WPT streams suite (which exercises the modified pull and cancel algorithms),streams.test.js, the TextEncoderStream/TextDecoderStream tests (they sharerunNativeArmand the release deferral), the node webstreams compression tests and the bun-types test, pass locally, the stream suites underBUN_JSC_validateExceptionChecks=1.httpserver, brotli body, 10 MiB guard) on a debug build: a 1 GiB bomb goes fromfirstChunk: 1073741824, +1013 MB RSS tofirstChunk: 65536, +58 MB; the 5 GiB / 4041-byte case that was OOM-killed reportsfirstChunk: 65536, +53 MB.Background
CompressionStream/DecompressionStreamis aJSTransformStreamsubclass whose transform and flush algorithms are C++ arms driving a Rust coder (one zlib/brotli/zstd context plus an output buffer) throughextern "C"calls. Chunks over 128 KiB run the coder on the thread pool and complete throughBun__CompressionStream__deliverAsync. The readable side queues one piece at a time (count-based high water mark of 1, as in Node and Chromium), so the option here is a byte bound on the pieces, not a queue length.[[backpressure]]flag; the readable's pull algorithm clears it when the consumer wants more, and a pending write waits for that. The pull algorithm is therefore exactly the "consumer has room" event, which is why a pending chunk is stepped from it. The transform algorithm's own promise is what keeps the writable's in-flight write pending.readStreamIntoSink(for example as aResponsebody inBun.serve), the arm writes coder output straight into the sink, which reports backpressure from the write and later calls onReady; the pump detaches the sink from the transform when it finishes. Those two callbacks are the sink-side equivalents of pull and cancel.ClearAlgorithmsis the shared terminal (after flush, on error, on cancel) and frees the coder eagerly instead of waiting for the cell's finalizer, deferring while an arm is on the stack or a pool step holds it; this change adds "a chunk is pending across turns" to that list.Earlier revision
The first revision registered a promise reaction on the readable's
backpressureChangePromise(or a gate promise in the sink-ready slot) to resume a pending chunk, which needed a reaction handler, the gate, and an extra unblock-write step in the cancel algorithm's early-return branch; the step sizes were constants (64 KiB on the JS thread, 1 MiB on the pool). Review asked for a configurable bound and less machinery, hence the current shape.