[Flight] Fix broken byte stream parsing caused by buffer detachment (#35127)
This PR fixes a critical bug where `ReadableStream({type: 'bytes'})` instances passed through React Server Components (RSC) would stall after reading only the first chunk or the first few chunks in the client. This issue was masked by using `web-streams-polyfill` in tests, but manifests with native Web Streams implementations. The root cause is that when a chunk is enqueued to a `ReadableByteStreamController`, the spec requires the underlying ArrayBuffer to be synchronously transferred/detached. In the React Flight Client's chunk parsing, embedded byte stream chunks are created as views into the incoming RSC stream chunk buffer using `new Uint8Array(chunk.buffer, offset, length)`. When embedded byte stream chunks are enqueued, they can detach the shared buffer, leaving the RSC stream parsing in a broken state. The fix is to copy embedded byte stream chunks before enqueueing them, preventing buffer detachment from affecting subsequent parsing. To not affect performance too much, we use a zero-copy optimization: when a chunk ends exactly at the end of the RSC stream chunk, or when the row spans into the next RSC chunk, no further parsing will access that buffer, so we can safely enqueue the view directly without copying. We now also enqueue embedded byte stream chunks immediately as they are parsed, without waiting for the full row to complete. To simplify the logic in the client, we introduce a new `'b'` protocol tag specifically for byte stream chunks. The server now emits `'b'` instead of `'o'` for `Uint8Array` chunks from byte streams (detected via `supportsBYOB`). This allows the client to recognize byte stream chunks without needing to track stream IDs. Tests now use the proper Jest environment with native Web Streams instead of polyfills, exposing and validating the fix for this issue.
Hendrik Liebau committed
Nov 13, 2025 at 21:23 UTC
93fc57400b0cf9a2bfdc2765d94ee9984f2fd97f
6 files changed
+202
-96
packages/react-client/src/ReactFlightClient.js
+45
-14
@@ -4857,6 +4857,7 @@ export function processBinaryChunk(
4857
resolvedRowTag === 65 /* "A" */ ||
4858
resolvedRowTag === 79 /* "O" */ ||
4859
resolvedRowTag === 111 /* "o" */ ||
4860
+ resolvedRowTag === 98 /* "b" */ ||
4861
resolvedRowTag === 85 /* "U" */ ||
4862
resolvedRowTag === 83 /* "S" */ ||
4863
resolvedRowTag === 115 /* "s" */ ||
@@ -4916,14 +4917,31 @@ export function processBinaryChunk(
4917
// We found the last chunk of the row
4918
const length = lastIdx - i;
4919
const lastChunk = new Uint8Array(chunk.buffer, offset, length);
4919
- processFullBinaryRow(
4920
- response,
4921
- streamState,
4922
- rowID,
4923
- rowTag,
4924
- buffer,
4925
- lastChunk,
4926
- );
4920
+
4921
+ // Check if this is a Uint8Array for a byte stream. We enqueue it
4922
+ // immediately but need to determine if we can use zero-copy or must copy.
4923
+ if (rowTag === 98 /* "b" */) {
4924
+ resolveBuffer(
4925
+ response,
4926
+ rowID,
4927
+ // If we're at the end of the RSC chunk, no more parsing will access
4928
+ // this buffer and we don't need to copy the chunk to allow detaching
4929
+ // the buffer, otherwise we need to copy.
4930
+ lastIdx === chunkLength ? lastChunk : lastChunk.slice(),
4931
+ streamState,
4932
+ );
4933
+ } else {
4934
+ // Process all other row types.
4935
+ processFullBinaryRow(
4936
+ response,
4937
+ streamState,
4938
+ rowID,
4939
+ rowTag,
4940
+ buffer,
4941
+ lastChunk,
4942
+ );
4943
+ }
4944
+
4945
// Reset state machine for a new row
4946
i = lastIdx;
4947
if (rowState === ROW_CHUNK_BY_NEWLINE) {
@@ -4936,14 +4954,27 @@ export function processBinaryChunk(
4954
rowLength = 0;
4955
buffer.length = 0;
4956
} else {
4939
- // The rest of this row is in a future chunk. We stash the rest of the
4940
- // current chunk until we can process the full row.
4957
+ // The rest of this row is in a future chunk.
4958
const length = chunk.byteLength - i;
4959
const remainingSlice = new Uint8Array(chunk.buffer, offset, length);
4943
- buffer.push(remainingSlice);
4944
- // Update how many bytes we're still waiting for. If we're looking for
4945
- // a newline, this doesn't hurt since we'll just ignore it.
4946
- rowLength -= remainingSlice.byteLength;
4960
+
4961
+ // For byte streams, we can enqueue the partial row immediately without
4962
+ // copying since we're at the end of the RSC chunk and no more parsing
4963
+ // will access this buffer.
4964
+ if (rowTag === 98 /* "b" */) {
4965
+ // Update how many bytes we're still waiting for. We need to do this
4966
+ // before enqueueing, as enqueue will detach the buffer and byteLength
4967
+ // will become 0.
4968
+ rowLength -= remainingSlice.byteLength;
4969
+ resolveBuffer(response, rowID, remainingSlice, streamState);
4970
+ } else {
4971
+ // For other row types, stash the rest of the current chunk until we can
4972
+ // process the full row.
4973
+ buffer.push(remainingSlice);
4974
+ // Update how many bytes we're still waiting for. If we're looking for
4975
+ // a newline, this doesn't hurt since we'll just ignore it.
4976
+ rowLength -= remainingSlice.byteLength;
4977
+ }
4978
break;
4979
}
4980
}
packages/react-server-dom-turbopack/src/__tests__/ReactFlightTurbopackDOMEdge-test.js
+1
-8
@@ -5,18 +5,11 @@
5
* LICENSE file in the root directory of this source tree.
6
*
7
* @emails react-core
8
+ * @jest-environment ./scripts/jest/ReactDOMServerIntegrationEnvironment
9
*/
10
11
'use strict';
12
12
-// Polyfills for test environment
13
-global.ReadableStream =
14
- require('web-streams-polyfill/ponyfill/es6').ReadableStream;
15
-global.WritableStream =
16
- require('web-streams-polyfill/ponyfill/es6').WritableStream;
17
-global.TextEncoder = require('util').TextEncoder;
18
-global.TextDecoder = require('util').TextDecoder;
19
-
13
let clientExports;
14
let turbopackMap;
15
let turbopackModules;
packages/react-server-dom-turbopack/src/__tests__/ReactFlightTurbopackDOMReplyEdge-test.js
+1
-8
@@ -5,17 +5,11 @@
5
* LICENSE file in the root directory of this source tree.
6
*
7
* @emails react-core
8
+ * @jest-environment ./scripts/jest/ReactDOMServerIntegrationEnvironment
9
*/
10
11
'use strict';
12
12
-// Polyfills for test environment
13
-global.ReadableStream =
14
- require('web-streams-polyfill/ponyfill/es6').ReadableStream;
15
-global.TextEncoder = require('util').TextEncoder;
16
-global.TextDecoder = require('util').TextDecoder;
17
-
18
-// let serverExports;
13
let turbopackServerMap;
14
let ReactServerDOMServer;
15
let ReactServerDOMClient;
@@ -29,7 +23,6 @@ describe('ReactFlightDOMTurbopackReply', () => {
23
require('react-server-dom-turbopack/server.edge'),
24
);
25
const TurbopackMock = require('./utils/TurbopackMock');
32
- // serverExports = TurbopackMock.serverExports;
26
turbopackServerMap = TurbopackMock.turbopackServerMap;
27
ReactServerDOMServer = require('react-server-dom-turbopack/server.edge');
28
jest.resetModules();
packages/react-server-dom-webpack/src/__tests__/ReactFlightDOMEdge-test.js
+120
-32
@@ -10,18 +10,6 @@
10
11
'use strict';
12
13
-// Polyfills for test environment
14
-global.ReadableStream =
15
- require('web-streams-polyfill/ponyfill/es6').ReadableStream;
16
-global.WritableStream =
17
- require('web-streams-polyfill/ponyfill/es6').WritableStream;
18
-global.TextEncoder = require('util').TextEncoder;
19
-global.TextDecoder = require('util').TextDecoder;
20
-global.Blob = require('buffer').Blob;
21
-if (typeof File === 'undefined' || typeof FormData === 'undefined') {
22
- global.File = require('buffer').File || require('undici').File;
23
- global.FormData = require('undici').FormData;
24
-}
13
// Patch for Edge environments for global scope
14
global.AsyncLocalStorage = require('async_hooks').AsyncLocalStorage;
15
@@ -127,8 +115,16 @@ describe('ReactFlightDOMEdge', () => {
115
chunk.set(prevChunk, 0);
116
chunk.set(value, prevChunk.length);
117
if (chunk.length > 50) {
118
+ // Copy the part we're keeping (prevChunk) to avoid buffer
119
+ // transfer. When we enqueue the partial chunk below, downstream
120
+ // consumers (like byte streams in the Flight Client) may detach
121
+ // the underlying buffer. Since prevChunk would share the same
122
+ // buffer, we copy it first so it has its own independent buffer.
123
+ // TODO: Should we just use {type: 'bytes'} for this stream to
124
+ // always transfer ownership, and not only "accidentally" when we
125
+ // enqueue in the Flight Client?
126
+ prevChunk = chunk.slice(chunk.length - 50);
127
controller.enqueue(chunk.subarray(0, chunk.length - 50));
131
- prevChunk = chunk.subarray(chunk.length - 50);
128
} else {
129
// Wait to see if we get some more bytes to join in.
130
prevChunk = chunk;
@@ -1118,25 +1114,121 @@ describe('ReactFlightDOMEdge', () => {
1114
expect(streamedBuffers).toEqual(buffers);
1115
});
1116
1117
+ it('should support binary ReadableStreams', async () => {
1118
+ const encoder = new TextEncoder();
1119
+ const words = ['Hello', 'streaming', 'world'];
1120
+
1121
+ const stream = new ReadableStream({
1122
+ type: 'bytes',
1123
+ async start(controller) {
1124
+ for (let i = 0; i < words.length; i++) {
1125
+ const chunk = encoder.encode(words[i] + ' ');
1126
+ controller.enqueue(chunk);
1127
+ }
1128
+ controller.close();
1129
+ },
1130
+ });
1131
+
1132
+ const rscStream = await serverAct(() =>
1133
+ ReactServerDOMServer.renderToReadableStream(stream, {}),
1134
+ );
1135
+
1136
+ const result = await ReactServerDOMClient.createFromReadableStream(
1137
+ rscStream,
1138
+ {
1139
+ serverConsumerManifest: {
1140
+ moduleMap: null,
1141
+ moduleLoading: null,
1142
+ },
1143
+ },
1144
+ );
1145
+
1146
+ const reader = result.getReader();
1147
+ const decoder = new TextDecoder();
1148
+
1149
+ let text = '';
1150
+ let entry;
1151
+ while (!(entry = await reader.read()).done) {
1152
+ text += decoder.decode(entry.value);
1153
+ }
1154
+
1155
+ expect(text).toBe('Hello streaming world ');
1156
+ });
1157
+
1158
+ it('should support large binary ReadableStreams', async () => {
1159
+ const chunkCount = 100;
1160
+ const chunkSize = 1024;
1161
+ const expectedBytes = [];
1162
+
1163
+ const stream = new ReadableStream({
1164
+ type: 'bytes',
1165
+ start(controller) {
1166
+ for (let i = 0; i < chunkCount; i++) {
1167
+ const chunk = new Uint8Array(chunkSize);
1168
+ for (let j = 0; j < chunkSize; j++) {
1169
+ chunk[j] = (i + j) % 256;
1170
+ }
1171
+ expectedBytes.push(...Array.from(chunk));
1172
+ controller.enqueue(chunk);
1173
+ }
1174
+ controller.close();
1175
+ },
1176
+ });
1177
+
1178
+ const rscStream = await serverAct(() =>
1179
+ ReactServerDOMServer.renderToReadableStream(stream, {}),
1180
+ );
1181
+
1182
+ const result = await ReactServerDOMClient.createFromReadableStream(
1183
+ // Use passThrough to split and rejoin chunks at arbitrary boundaries.
1184
+ passThrough(rscStream),
1185
+ {
1186
+ serverConsumerManifest: {
1187
+ moduleMap: null,
1188
+ moduleLoading: null,
1189
+ },
1190
+ },
1191
+ );
1192
+
1193
+ const reader = result.getReader();
1194
+ const receivedBytes = [];
1195
+ let entry;
1196
+ while (!(entry = await reader.read()).done) {
1197
+ expect(entry.value instanceof Uint8Array).toBe(true);
1198
+ receivedBytes.push(...Array.from(entry.value));
1199
+ }
1200
+
1201
+ expect(receivedBytes).toEqual(expectedBytes);
1202
+ });
1203
+
1204
it('should support BYOB binary ReadableStreams', async () => {
1122
- const buffer = new Uint8Array([
1205
+ const sourceBytes = [
1206
123, 4, 10, 5, 100, 255, 244, 45, 56, 67, 43, 124, 67, 89, 100, 20,
1124
- ]).buffer;
1207
+ ];
1208
+
1209
+ // Create separate buffers for each typed array to avoid ArrayBuffer
1210
+ // transfer issues. Each view needs its own buffer because enqueue()
1211
+ // transfers ownership.
1212
const buffers = [
1126
- new Int8Array(buffer, 1),
1127
- new Uint8Array(buffer, 2),
1128
- new Uint8ClampedArray(buffer, 2),
1129
- new Int16Array(buffer, 2),
1130
- new Uint16Array(buffer, 2),
1131
- new Int32Array(buffer, 4),
1132
- new Uint32Array(buffer, 4),
1133
- new Float32Array(buffer, 4),
1134
- new Float64Array(buffer, 0),
1135
- new BigInt64Array(buffer, 0),
1136
- new BigUint64Array(buffer, 0),
1137
- new DataView(buffer, 3),
1213
+ new Int8Array(sourceBytes.slice(1)),
1214
+ new Uint8Array(sourceBytes.slice(2)),
1215
+ new Uint8ClampedArray(sourceBytes.slice(2)),
1216
+ new Int16Array(new Uint8Array(sourceBytes.slice(2)).buffer),
1217
+ new Uint16Array(new Uint8Array(sourceBytes.slice(2)).buffer),
1218
+ new Int32Array(new Uint8Array(sourceBytes.slice(4)).buffer),
1219
+ new Uint32Array(new Uint8Array(sourceBytes.slice(4)).buffer),
1220
+ new Float32Array(new Uint8Array(sourceBytes.slice(4)).buffer),
1221
+ new Float64Array(new Uint8Array(sourceBytes.slice(0)).buffer),
1222
+ new BigInt64Array(new Uint8Array(sourceBytes.slice(0)).buffer),
1223
+ new BigUint64Array(new Uint8Array(sourceBytes.slice(0)).buffer),
1224
+ new DataView(new Uint8Array(sourceBytes.slice(3)).buffer),
1225
];
1226
1227
+ // Save expected bytes before enqueueing (which will detach the buffers).
1228
+ const expectedBytes = buffers.flatMap(c =>
1229
+ Array.from(new Uint8Array(c.buffer, c.byteOffset, c.byteLength)),
1230
+ );
1231
+
1232
// This a binary stream where each chunk ends up as Uint8Array.
1233
const s = new ReadableStream({
1234
type: 'bytes',
@@ -1176,11 +1268,7 @@ describe('ReactFlightDOMEdge', () => {
1268
1269
// The streamed buffers might be in different chunks and in Uint8Array form but
1270
// the concatenated bytes should be the same.
1179
- expect(streamedBuffers.flatMap(t => Array.from(t))).toEqual(
1180
- buffers.flatMap(c =>
1181
- Array.from(new Uint8Array(c.buffer, c.byteOffset, c.byteLength)),
1182
- ),
1183
- );
1271
+ expect(streamedBuffers.flatMap(t => Array.from(t))).toEqual(expectedBytes);
1272
});
1273
1274
// @gate !__DEV__ || enableComponentPerformanceTrack
packages/react-server-dom-webpack/src/__tests__/ReactFlightDOMReplyEdge-test.js
+24
-31
@@ -10,18 +10,6 @@
10
11
'use strict';
12
13
-// Polyfills for test environment
14
-global.ReadableStream =
15
- require('web-streams-polyfill/ponyfill/es6').ReadableStream;
16
-global.TextEncoder = require('util').TextEncoder;
17
-global.TextDecoder = require('util').TextDecoder;
18
-
19
-global.Blob = require('buffer').Blob;
20
-if (typeof File === 'undefined' || typeof FormData === 'undefined') {
21
- global.File = require('buffer').File || require('undici').File;
22
- global.FormData = require('undici').FormData;
23
-}
24
-
13
let serverExports;
14
let webpackServerMap;
15
let ReactServerDOMServer;
@@ -194,24 +182,33 @@ describe('ReactFlightDOMReplyEdge', () => {
182
});
183
184
it('should support BYOB binary ReadableStreams', async () => {
197
- const buffer = new Uint8Array([
185
+ const sourceBytes = [
186
123, 4, 10, 5, 100, 255, 244, 45, 56, 67, 43, 124, 67, 89, 100, 20,
199
- ]).buffer;
187
+ ];
188
+
189
+ // Create separate buffers for each typed array to avoid ArrayBuffer
190
+ // transfer issues. Each view needs its own buffer because enqueue()
191
+ // transfers ownership.
192
const buffers = [
201
- new Int8Array(buffer, 1),
202
- new Uint8Array(buffer, 2),
203
- new Uint8ClampedArray(buffer, 2),
204
- new Int16Array(buffer, 2),
205
- new Uint16Array(buffer, 2),
206
- new Int32Array(buffer, 4),
207
- new Uint32Array(buffer, 4),
208
- new Float32Array(buffer, 4),
209
- new Float64Array(buffer, 0),
210
- new BigInt64Array(buffer, 0),
211
- new BigUint64Array(buffer, 0),
212
- new DataView(buffer, 3),
193
+ new Int8Array(sourceBytes.slice(1)),
194
+ new Uint8Array(sourceBytes.slice(2)),
195
+ new Uint8ClampedArray(sourceBytes.slice(2)),
196
+ new Int16Array(new Uint8Array(sourceBytes.slice(2)).buffer),
197
+ new Uint16Array(new Uint8Array(sourceBytes.slice(2)).buffer),
198
+ new Int32Array(new Uint8Array(sourceBytes.slice(4)).buffer),
199
+ new Uint32Array(new Uint8Array(sourceBytes.slice(4)).buffer),
200
+ new Float32Array(new Uint8Array(sourceBytes.slice(4)).buffer),
201
+ new Float64Array(new Uint8Array(sourceBytes.slice(0)).buffer),
202
+ new BigInt64Array(new Uint8Array(sourceBytes.slice(0)).buffer),
203
+ new BigUint64Array(new Uint8Array(sourceBytes.slice(0)).buffer),
204
+ new DataView(new Uint8Array(sourceBytes.slice(3)).buffer),
205
];
206
207
+ // Save expected bytes before enqueueing (which will detach the buffers).
208
+ const expectedBytes = buffers.flatMap(c =>
209
+ Array.from(new Uint8Array(c.buffer, c.byteOffset, c.byteLength)),
210
+ );
211
+
212
// This a binary stream where each chunk ends up as Uint8Array.
213
const s = new ReadableStream({
214
type: 'bytes',
@@ -239,11 +236,7 @@ describe('ReactFlightDOMReplyEdge', () => {
236
237
// The streamed buffers might be in different chunks and in Uint8Array form but
238
// the concatenated bytes should be the same.
242
- expect(streamedBuffers.flatMap(t => Array.from(t))).toEqual(
243
- buffers.flatMap(c =>
244
- Array.from(new Uint8Array(c.buffer, c.byteOffset, c.byteLength)),
245
- ),
246
- );
239
+ expect(streamedBuffers.flatMap(t => Array.from(t))).toEqual(expectedBytes);
240
});
241
242
it('should abort when parsing an incomplete payload', async () => {
packages/react-server/src/ReactFlightServer.js
+11
-3
@@ -1149,6 +1149,8 @@ function serializeReadableStream(
1149
supportsBYOB = false;
1150
}
1151
}
1152
+ // At this point supportsBYOB is guaranteed to be a boolean.
1153
+ const isByteStream: boolean = supportsBYOB;
1154
1155
const reader = stream.getReader();
1156
@@ -1172,7 +1174,7 @@ function serializeReadableStream(
1174
// The task represents the Stop row. This adds a Start row.
1175
request.pendingChunks++;
1176
const startStreamRow =
1175
- streamTask.id.toString(16) + ':' + (supportsBYOB ? 'r' : 'R') + '\n';
1177
+ streamTask.id.toString(16) + ':' + (isByteStream ? 'r' : 'R') + '\n';
1178
request.completedRegularChunks.push(stringToChunk(startStreamRow));
1179
1180
function progress(entry: {done: boolean, value: ReactClientValue, ...}) {
@@ -1190,9 +1192,15 @@ function serializeReadableStream(
1192
callOnAllReadyIfReady(request);
1193
} else {
1194
try {
1193
- streamTask.model = entry.value;
1195
request.pendingChunks++;
1195
- tryStreamTask(request, streamTask);
1196
+ streamTask.model = entry.value;
1197
+ if (isByteStream) {
1198
+ // Chunks of byte streams are always Uint8Array instances.
1199
+ const chunk: Uint8Array = (streamTask.model: any);
1200
+ emitTypedArrayChunk(request, streamTask.id, 'b', chunk, false);
1201
+ } else {
1202
+ tryStreamTask(request, streamTask);
1203
+ }
1204
enqueueFlush(request);
1205
reader.read().then(progress, error);
1206
} catch (x) {