@samitouri / QOS-React-1 / commits / c014813413

[Flight] Fix stranded row content under Node stream backpressure (#36516)

The Flight Server emits Text and TypedArray rows as two chunks: a header that gives the row's id, type, and content length, followed by the content itself. These two chunks were pushed into `completedRegularChunks` (and `completedDebugChunks` in DEV) as separate items, so when the destination signaled backpressure between them, the flush would write the header and then break out before reaching the content. The content chunk was left stranded at the head of the queue. Async work running while the destination was paused appended new rows to `completedImportChunks` / `completedHintChunks`, and the next drain flushed those queues first — splicing the newly-arrived bytes into the position the Flight Client expects to read as the original row's content. From there the Flight Client read rows from the wrong byte offsets and the model failed to deserialize. This only surfaced on the Node stream path. `createFakeWritableFromReadableStreamController`, used by `renderToReadableStream`, always returns `true` from `write()`, so the flush loop never saw backpressure. The fix pushes a `NEXT_TWO_CHUNKS_ARE_ATOMIC` sentinel ahead of each `headerChunk` / `contentChunk` pair in `completedRegularChunks` and `completedDebugChunks`. The flush loops detect the sentinel and write the two chunks that follow it together before re-checking backpressure, so backpressure can still break between rows but never within one. ### Alternatives considered - **Pushing the pair as a `[headerChunk, contentChunk]` tuple.** Simpler but allocates an array per row. The required `isArray` branch in the flush hot path is likely comparable with the symbol check. This also violates the opaque `Chunk` type boundary. - **Concatenating header and content into one chunk.** Bad for memory — typed-array content can be large. - **Storing atomic groups in a separate queue.** Conceptually wrong and risks breaking reference-ordering assumptions between rows. - **Ignoring backpressure until the regulars queue is empty.** Defeats the point of backpressure. - **Wrapping the tuple behind a host-config API** (`writeAdjacentChunks` / `isAdjacentChunks` / `chunksToAdjacentChunks` / `getAdjacentChunksLength`). Keeps the implementation opaque but adds four exports per host config. Also has the tuple overhead. - **Begin/end sentinels for variable-length atomic groups.** Not needed — only Text and TypedArray rows use this pattern, and both are pairs.

Hendrik Liebau committed May 27, 2026 at 22:19 UTC c014813413dff8cc159180c77d81eaeb58a1c527

3 files changed +279 -18

packages/react-server-dom-webpack/src/__tests__/ReactFlightDOMNode-test.js

+153

@@ -2092,4 +2092,157 @@ describe('ReactFlightDOMNode', () => {

2092 globalThis.eval = previousEval;

2093 }

2094 });

2095 +

2096 + // Shared scenario for the two regression tests below. Both guard against

2097 + // the same destination-backpressure bug — emitTextChunk and

2098 + // emitTypedArrayChunk each push a [headerChunk, contentChunk] pair into

2099 + // completedRegularChunks. Before the fix, a flush that broke between the

2100 + // two writes left the content chunk stranded at the head of the queue,

2101 + // and the next flush emitted any newly-arrived Import rows ahead of it —

2102 + // splicing Import bytes into the position the Flight Client expects to

2103 + // read as the row's content.

2104 + //

2105 + // The scenario embeds `payload` in the model under the key `payload` and

2106 + // returns the deserialized model. Each test asserts that result.payload

2107 + // round-trips identically to what the Flight Server emitted.

2108 + async function runScenarioWithBackpressureBetweenHeaderAndContent(payload) {

2109 + function Client1() {

2110 + return <span>client1</span>;

2111 + }

2112 + // Client1's Import row must exceed VIEW_SIZE (4096) so writeStringChunk

2113 + // takes its BIG path and calls destination.write directly. That write

2114 + // returning false is what triggers the backpressure we want to test.

2115 + const Client1Reference = clientExports(

2116 + Client1,

2117 + 1,

2118 + '/' + 'a'.repeat(5000),

2119 + Promise.resolve(),

2120 + );

2121 +

2122 + function Client2() {

2123 + return <span>client2</span>;

2124 + }

2125 + const Client2Reference = clientExports(

2126 + Client2,

2127 + 2,

2128 + '/client2.js',

2129 + Promise.resolve(),

2130 + );

2131 +

2132 + let resolveAsync;

2133 + const asyncPromise = new Promise(resolve => {

2134 + resolveAsync = resolve;

2135 + });

2136 +

2137 + async function AsyncWrapper() {

2138 + await asyncPromise;

2139 + return <Client2Reference />;

2140 + }

2141 +

2142 + const model = {

2143 + client: <Client1Reference />,

2144 + payload,

2145 + async: <AsyncWrapper />,

2146 + };

2147 +

2148 + const heldCallbacks = [];

2149 + const collectedChunks = [];

2150 +

2151 + // A destination that returns false from every write (highWaterMark: 1 in

2152 + // byte mode) and never completes any of them until the test releases the

2153 + // stored callback. This gives us deterministic control over when each write

2154 + // finishes and when 'drain' fires.

2155 + const destination = new Stream.Writable({

2156 + highWaterMark: 1,

2157 + write(chunk, encoding, callback) {

2158 + collectedChunks.push(

2159 + Buffer.isBuffer(chunk) ? chunk : Buffer.from(chunk, encoding),

2160 + );

2161 + heldCallbacks.push(callback);

2162 + },

2163 + });

2164 +

2165 + const finished = new Promise((resolve, reject) => {

2166 + destination.on('finish', resolve);

2167 + destination.on('error', reject);

2168 + });

2169 +

2170 + // First flush: Client1's huge Import row hits backpressure, so the flush

2171 + // loop reaches the payload row's [headerChunk, contentChunk] pair while

2172 + // destinationHasCapacity is already false. Before the fix, this would

2173 + // have encoded just the header into currentView, broken the loop, and

2174 + // let completeWriting flush the header as its own write — stranding

2175 + // the content chunk at the front of completedRegularChunks.

2176 + const {pipe} = await serverAct(() =>

2177 + ReactServerDOMServer.renderToPipeableStream(model, webpackMap),

2178 + );

2179 + await serverAct(() => {

2180 + pipe(destination);

2181 + });

2182 +

2183 + // While the destination is still paused, push Client2's Import row into

2184 + // completedImportChunks. No flush runs (request.destination is null after

2185 + // the first flush's backpressure break).

2186 + await serverAct(() => {

2187 + resolveAsync();

2188 + });

2189 +

2190 + // Release callbacks one at a time. The drain that empties the writable

2191 + // buffer triggers flushCompletedChunks; before the fix, this is where

2192 + // Client2's newly-queued Import row would have been emitted ahead of

2193 + // the still-orphaned payload content chunk.

2194 + while (heldCallbacks.length > 0) {

2195 + await serverAct(() => {

2196 + const cb = heldCallbacks.shift();

2197 + cb();

2198 + });

2199 + }

2200 +

2201 + await finished;

2202 +

2203 + const readable = new Stream.Readable({read() {}});

2204 + for (let i = 0; i < collectedChunks.length; i++) {

2205 + readable.push(collectedChunks[i]);

2206 + }

2207 + readable.push(null);

2208 +

2209 + const response = ReactServerDOMClient.createFromNodeStream(readable, {

2210 + moduleMap: null,

2211 + moduleLoading: null,

2212 + });

2213 + return await response;

2214 + }

2215 +

2216 + it("keeps a Text row's header and content chunks adjacent when a flush hits backpressure between them", async () => {

2217 + // length >= 1024 makes the Flight Server outline this as a Text row via

2218 + // serializeLargeTextString, which is where emitTextChunk pushes its

2219 + // [headerChunk, textChunk] pair.

2220 + const largeText = 'x'.repeat(2048);

2221 +

2222 + const result =

2223 + await runScenarioWithBackpressureBetweenHeaderAndContent(largeText);

2224 +

2225 + // Before the fix, the Flight Client would have framed Client2's Import

2226 + // row bytes as text-row content, making result.payload `<id>:I[...]...`

2227 + // garbage rather than the x's the Flight Server emitted.

2228 + expect(result.payload).toBe(largeText);

2229 + });

2230 +

2231 + it("keeps a TypedArray row's header and content chunks adjacent when a flush hits backpressure between them", async () => {

2232 + // emitTypedArrayChunk pushes the same [headerChunk, contentChunk] pair as

2233 + // emitTextChunk. Before the fix, a flush break after the header would have

2234 + // stranded the content chunk in exactly the same way.

2235 + const binaryData = new Uint8Array(1024);

2236 + for (let i = 0; i < binaryData.length; i++) {

2237 + binaryData[i] = i % 256;

2238 + }

2239 +

2240 + const result =

2241 + await runScenarioWithBackpressureBetweenHeaderAndContent(binaryData);

2242 +

2243 + // Before the fix, the typed array's bytes would have been replaced by

2244 + // Client2's Import row bytes followed by whatever happened to land in

2245 + // the next 1024-byte window.

2246 + expect(result.payload).toEqual(binaryData);

2247 + });

2248 });

packages/react-server/src/ReactFlightServer.js

+124 -17

@@ -23,6 +23,7 @@ import {

23 scheduleMicrotask,

24 flushBuffered,

25 beginWriting,

26 + writeChunk,

27 writeChunkAndReturn,

28 stringToChunk,

29 typedArrayToBinaryChunk,

@@ -560,6 +561,12 @@ const CLOSED = 14;

561 const RENDER = 20;

562 const PRERENDER = 21;

563

564 +// Marker pushed before a [headerChunk, contentChunk] pair in

565 +// completedRegularChunks / completedDebugChunks to signal that the next two

566 +// entries must be written atomically — see emitTextChunk and

567 +// emitTypedArrayChunk for why, and flushCompletedChunks for how it's read.

568 +const NEXT_TWO_CHUNKS_ARE_ATOMIC: symbol = Symbol();

569 +

570 export type Request = {

571 status: 10 | 11 | 12 | 13 | 14,

572 type: 20 | 21,

@@ -576,7 +583,13 @@ export type Request = {

583 pingedTasks: Array<Task>,

584 completedImportChunks: Array<Chunk>,

585 completedHintChunks: Array<Chunk>,

579 - completedRegularChunks: Array<Chunk | BinaryChunk>,

586 + // Text and TypedArray rows are pushed as a NEXT_TWO_CHUNKS_ARE_ATOMIC

587 + // sentinel followed by their [headerChunk, contentChunk] pair, so that

588 + // flushCompletedChunks can write the pair atomically and never strand the

589 + // content chunk on a backpressure break.

590 + completedRegularChunks: Array<

591 + Chunk | BinaryChunk | typeof NEXT_TWO_CHUNKS_ARE_ATOMIC,

592 + >,

593 completedErrorChunks: Array<Chunk>,

594 writtenSymbols: Map<symbol, number>,

595 writtenClientReferences: Map<ClientReferenceKey, number>,

@@ -594,7 +607,11 @@ export type Request = {

607 abortTime: number,

608 // DEV-only

609 pendingDebugChunks: number,

597 - completedDebugChunks: Array<Chunk | BinaryChunk>,

610 + // See completedRegularChunks for why some entries are preceded by the

611 + // NEXT_TWO_CHUNKS_ARE_ATOMIC sentinel.

612 + completedDebugChunks: Array<

613 + Chunk | BinaryChunk | typeof NEXT_TWO_CHUNKS_ARE_ATOMIC,

614 + >,

615 debugDestination: null | Destination,

616 environmentName: () => string,

617 filterStackFrame: (

@@ -695,7 +712,9 @@ function RequestInstance(

712 this.pingedTasks = pingedTasks;

713 this.completedImportChunks = ([]: Array<Chunk>);

714 this.completedHintChunks = ([]: Array<Chunk>);

698 - this.completedRegularChunks = ([]: Array<Chunk | BinaryChunk>);

715 + this.completedRegularChunks = ([]: Array<

716 + Chunk | BinaryChunk | typeof NEXT_TWO_CHUNKS_ARE_ATOMIC,

717 + >);

718 this.completedErrorChunks = ([]: Array<Chunk>);

719 this.writtenSymbols = new Map();

720 this.writtenClientReferences = new Map();

@@ -711,7 +730,9 @@ function RequestInstance(

730

731 if (__DEV__) {

732 this.pendingDebugChunks = 0;

714 - this.completedDebugChunks = ([]: Array<Chunk>);

733 + this.completedDebugChunks = ([]: Array<

734 + Chunk | BinaryChunk | typeof NEXT_TWO_CHUNKS_ARE_ATOMIC,

735 + >);

736 this.debugDestination = null;

737 this.environmentName =

738 environmentName === undefined

@@ -4691,10 +4712,25 @@ function emitTypedArrayChunk(

4712 const binaryLength = byteLengthOfBinaryChunk(binaryChunk);

4713 const row = id.toString(16) + ':' + tag + binaryLength.toString(16) + ',';

4714 const headerChunk = stringToChunk(row);

4715 + // Push a NEXT_TWO_CHUNKS_ARE_ATOMIC sentinel before the header so that

4716 + // flushCompletedChunks can write the header and binary chunks atomically.

4717 + // Otherwise, if the destination's backpressure flips between the two writes,

4718 + // the content chunk would be stranded at the front of the queue and the next

4719 + // drain would emit Import or Hint chunks between the header and the content —

4720 + // and the Flight Client would frame those intervening bytes as this row's

4721 + // content.

4722 if (__DEV__ && debug) {

4695 - request.completedDebugChunks.push(headerChunk, binaryChunk);

4723 + request.completedDebugChunks.push(

4724 + NEXT_TWO_CHUNKS_ARE_ATOMIC,

4725 + headerChunk,

4726 + binaryChunk,

4727 + );

4728 } else {

4697 - request.completedRegularChunks.push(headerChunk, binaryChunk);

4729 + request.completedRegularChunks.push(

4730 + NEXT_TWO_CHUNKS_ARE_ATOMIC,

4731 + headerChunk,

4732 + binaryChunk,

4733 + );

4734 }

4735 }

4736

@@ -4719,10 +4755,19 @@ function emitTextChunk(

4755 const binaryLength = byteLengthOfChunk(textChunk);

4756 const row = id.toString(16) + ':T' + binaryLength.toString(16) + ',';

4757 const headerChunk = stringToChunk(row);

4758 + // See emitTypedArrayChunk for why the pair is preceded by a sentinel.

4759 if (__DEV__ && debug) {

4723 - request.completedDebugChunks.push(headerChunk, textChunk);

4760 + request.completedDebugChunks.push(

4761 + NEXT_TWO_CHUNKS_ARE_ATOMIC,

4762 + headerChunk,

4763 + textChunk,

4764 + );

4765 } else {

4725 - request.completedRegularChunks.push(headerChunk, textChunk);

4766 + request.completedRegularChunks.push(

4767 + NEXT_TWO_CHUNKS_ARE_ATOMIC,

4768 + headerChunk,

4769 + textChunk,

4770 + );

4771 }

4772 }

4773

@@ -6014,9 +6059,27 @@ function flushCompletedChunks(request: Request): void {

6059 const debugChunks = request.completedDebugChunks;

6060 let i = 0;

6061 for (; i < debugChunks.length; i++) {

6017 - request.pendingDebugChunks--;

6018 - const chunk = debugChunks[i];

6019 - writeChunkAndReturn(debugDestination, chunk);

6062 + const item = debugChunks[i];

6063 + if (item === NEXT_TWO_CHUNKS_ARE_ATOMIC) {

6064 + if (i + 2 >= debugChunks.length) {

6065 + throw new Error(

6066 + 'A chunk pair is incomplete. This is a bug in React.',

6067 + );

6068 + }

6069 + request.pendingDebugChunks -= 2;

6070 + writeChunk(

6071 + debugDestination,

6072 + ((debugChunks[i + 1]: any): Chunk | BinaryChunk),

6073 + );

6074 + writeChunk(

6075 + debugDestination,

6076 + ((debugChunks[i + 2]: any): Chunk | BinaryChunk),

6077 + );

6078 + i += 2;

6079 + } else {

6080 + request.pendingDebugChunks--;

6081 + writeChunk(debugDestination, ((item: any): Chunk | BinaryChunk));

6082 + }

6083 }

6084 debugChunks.splice(0, i);

6085 } finally {

@@ -6064,9 +6127,31 @@ function flushCompletedChunks(request: Request): void {

6127 const debugChunks = request.completedDebugChunks;

6128 i = 0;

6129 for (; i < debugChunks.length; i++) {

6067 - request.pendingDebugChunks--;

6068 - const chunk = debugChunks[i];

6069 - const keepWriting: boolean = writeChunkAndReturn(destination, chunk);

6130 + const item = debugChunks[i];

6131 + let keepWriting: boolean;

6132 + if (item === NEXT_TWO_CHUNKS_ARE_ATOMIC) {

6133 + if (i + 2 >= debugChunks.length) {

6134 + throw new Error(

6135 + 'A chunk pair is incomplete. This is a bug in React.',

6136 + );

6137 + }

6138 + request.pendingDebugChunks -= 2;

6139 + writeChunk(

6140 + destination,

6141 + ((debugChunks[i + 1]: any): Chunk | BinaryChunk),

6142 + );

6143 + keepWriting = writeChunkAndReturn(

6144 + destination,

6145 + ((debugChunks[i + 2]: any): Chunk | BinaryChunk),

6146 + );

6147 + i += 2;

6148 + } else {

6149 + request.pendingDebugChunks--;

6150 + keepWriting = writeChunkAndReturn(

6151 + destination,

6152 + ((item: any): Chunk | BinaryChunk),

6153 + );

6154 + }

6155 if (!keepWriting) {

6156 request.destination = null;

6157 i++;

@@ -6080,9 +6165,31 @@ function flushCompletedChunks(request: Request): void {

6165 const regularChunks = request.completedRegularChunks;

6166 i = 0;

6167 for (; i < regularChunks.length; i++) {

6083 - request.pendingChunks--;

6084 - const chunk = regularChunks[i];

6085 - const keepWriting: boolean = writeChunkAndReturn(destination, chunk);

6168 + const item = regularChunks[i];

6169 + let keepWriting: boolean;

6170 + if (item === NEXT_TWO_CHUNKS_ARE_ATOMIC) {

6171 + if (i + 2 >= regularChunks.length) {

6172 + throw new Error(

6173 + 'A chunk pair is incomplete. This is a bug in React.',

6174 + );

6175 + }

6176 + request.pendingChunks -= 2;

6177 + writeChunk(

6178 + destination,

6179 + ((regularChunks[i + 1]: any): Chunk | BinaryChunk),

6180 + );

6181 + keepWriting = writeChunkAndReturn(

6182 + destination,

6183 + ((regularChunks[i + 2]: any): Chunk | BinaryChunk),

6184 + );

6185 + i += 2;

6186 + } else {

6187 + request.pendingChunks--;

6188 + keepWriting = writeChunkAndReturn(

6189 + destination,

6190 + ((item: any): Chunk | BinaryChunk),

6191 + );

6192 + }

6193 if (!keepWriting) {

6194 request.destination = null;

6195 i++;

scripts/error-codes/codes.json

+2 -1

@@ -585,5 +585,6 @@

585 "597": "The module \"%s\" is marked as an async ESM module but was loaded as a CJS proxy. This is probably a bug in the React Server Components bundler.",

586 "598": "Maximum update depth exceeded. This could be an infinite loop. This can happen when a component repeatedly calls setState during render phase or inside useLayoutEffect, causing infinite render loop. React limits the number of nested updates to prevent infinite loops.",

587 "599": "Expected an initialized chunk but got an initialized stream chunk instead. This payload may have been submitted by an older version of React.",

588 - "600": "A rejected Promise was passed to React without a `reason` property. React threw a generic error from where the Promise was used to assist in identifying the problematic Promise. Make sure that instrumented Promises correctly set the `reason` property when setting `status` to `'rejected'`."

588 + "600": "A rejected Promise was passed to React without a `reason` property. React threw a generic error from where the Promise was used to assist in identifying the problematic Promise. Make sure that instrumented Promises correctly set the `reason` property when setting `status` to `'rejected'`.",

589 + "601": "A chunk pair is incomplete. This is a bug in React."

590 }