[Flight] Wait for both streams to end before closing the response (#34301)
When a debug channel is defined, we must ensure that we don't close the Flight Client's response when the debug channel's readable is done, but the RSC stream is still flowing. Now, we wait for both streams to end before closing the response.
Hendrik Liebau committed
Aug 26, 2025 at 17:15 UTC
cacc20e37c9ec320b5d2aa13f86cfc999d269d6b
17 files changed
+514
-149
packages/react-server-dom-esm/src/client/ReactFlightDOMClientBrowser.js
+30
-13
@@ -101,6 +101,7 @@ function createResponseFromOptions(options: void | Options) {
101
function startReadingFromUniversalStream(
102
response: FlightResponse,
103
stream: ReadableStream,
104
+ onDone: () => void,
105
): void {
106
// This is the same as startReadingFromStream except this allows WebSocketStreams which
107
// return ArrayBuffer and string chunks instead of Uint8Array chunks. We could potentially
@@ -116,8 +117,7 @@ function startReadingFromUniversalStream(
117
...
118
}): void | Promise<void> {
119
if (done) {
119
- close(response);
120
- return;
120
+ return onDone();
121
}
122
if (value instanceof ArrayBuffer) {
123
// WebSockets can produce ArrayBuffer values in ReadableStreams.
@@ -139,7 +139,7 @@ function startReadingFromUniversalStream(
139
function startReadingFromStream(
140
response: FlightResponse,
141
stream: ReadableStream,
142
- isSecondaryStream: boolean,
142
+ onDone: () => void,
143
): void {
144
const streamState = createStreamState();
145
const reader = stream.getReader();
@@ -152,11 +152,7 @@ function startReadingFromStream(
152
...
153
}): void | Promise<void> {
154
if (done) {
155
- // If we're the secondary stream, then we don't close the response until the debug channel closes.
156
- if (!isSecondaryStream) {
157
- close(response);
158
- }
159
- return;
155
+ return onDone();
156
}
157
const buffer: Uint8Array = (value: any);
158
processBinaryChunk(response, streamState, buffer);
@@ -178,10 +174,20 @@ function createFromReadableStream<T>(
174
options.debugChannel &&
175
options.debugChannel.readable
176
) {
181
- startReadingFromUniversalStream(response, options.debugChannel.readable);
182
- startReadingFromStream(response, stream, true);
177
+ let streamDoneCount = 0;
178
+ const handleDone = () => {
179
+ if (++streamDoneCount === 2) {
180
+ close(response);
181
+ }
182
+ };
183
+ startReadingFromUniversalStream(
184
+ response,
185
+ options.debugChannel.readable,
186
+ handleDone,
187
+ );
188
+ startReadingFromStream(response, stream, handleDone);
189
} else {
184
- startReadingFromStream(response, stream, false);
190
+ startReadingFromStream(response, stream, close.bind(null, response));
191
}
192
return getRoot(response);
193
}
@@ -199,13 +205,24 @@ function createFromFetch<T>(
205
options.debugChannel &&
206
options.debugChannel.readable
207
) {
208
+ let streamDoneCount = 0;
209
+ const handleDone = () => {
210
+ if (++streamDoneCount === 2) {
211
+ close(response);
212
+ }
213
+ };
214
startReadingFromUniversalStream(
215
response,
216
options.debugChannel.readable,
217
+ handleDone,
218
);
206
- startReadingFromStream(response, (r.body: any), true);
219
+ startReadingFromStream(response, (r.body: any), handleDone);
220
} else {
208
- startReadingFromStream(response, (r.body: any), false);
221
+ startReadingFromStream(
222
+ response,
223
+ (r.body: any),
224
+ close.bind(null, response),
225
+ );
226
}
227
},
228
function (e) {
packages/react-server-dom-esm/src/client/ReactFlightDOMClientNode.js
+11
-11
@@ -63,7 +63,7 @@ export type Options = {
63
function startReadingFromStream(
64
response: Response,
65
stream: Readable,
66
- isSecondaryStream: boolean,
66
+ onEnd: () => void,
67
): void {
68
const streamState = createStreamState();
69
@@ -79,13 +79,7 @@ function startReadingFromStream(
79
reportGlobalError(response, error);
80
});
81
82
- stream.on('end', () => {
83
- // If we're the secondary stream, then we don't close the response until the
84
- // debug channel closes.
85
- if (!isSecondaryStream) {
86
- close(response);
87
- }
88
- });
82
+ stream.on('end', onEnd);
83
}
84
85
function createFromNodeStream<T>(
@@ -112,10 +106,16 @@ function createFromNodeStream<T>(
106
);
107
108
if (__DEV__ && options && options.debugChannel) {
115
- startReadingFromStream(response, options.debugChannel, false);
116
- startReadingFromStream(response, stream, true);
109
+ let streamEndedCount = 0;
110
+ const handleEnd = () => {
111
+ if (++streamEndedCount === 2) {
112
+ close(response);
113
+ }
114
+ };
115
+ startReadingFromStream(response, options.debugChannel, handleEnd);
116
+ startReadingFromStream(response, stream, handleEnd);
117
} else {
118
- startReadingFromStream(response, stream, false);
118
+ startReadingFromStream(response, stream, close.bind(null, response));
119
}
120
121
return getRoot(response);
packages/react-server-dom-parcel/src/client/ReactFlightDOMClientBrowser.js
+30
-13
@@ -102,6 +102,7 @@ function createDebugCallbackFromWritableStream(
102
function startReadingFromUniversalStream(
103
response: FlightResponse,
104
stream: ReadableStream,
105
+ onDone: () => void,
106
): void {
107
// This is the same as startReadingFromStream except this allows WebSocketStreams which
108
// return ArrayBuffer and string chunks instead of Uint8Array chunks. We could potentially
@@ -117,8 +118,7 @@ function startReadingFromUniversalStream(
118
...
119
}): void | Promise<void> {
120
if (done) {
120
- close(response);
121
- return;
121
+ return onDone();
122
}
123
if (value instanceof ArrayBuffer) {
124
// WebSockets can produce ArrayBuffer values in ReadableStreams.
@@ -140,7 +140,7 @@ function startReadingFromUniversalStream(
140
function startReadingFromStream(
141
response: FlightResponse,
142
stream: ReadableStream,
143
- isSecondaryStream: boolean,
143
+ onDone: () => void,
144
): void {
145
const streamState = createStreamState();
146
const reader = stream.getReader();
@@ -153,11 +153,7 @@ function startReadingFromStream(
153
...
154
}): void | Promise<void> {
155
if (done) {
156
- // If we're the secondary stream, then we don't close the response until the debug channel closes.
157
- if (!isSecondaryStream) {
158
- close(response);
159
- }
160
- return;
156
+ return onDone();
157
}
158
const buffer: Uint8Array = (value: any);
159
processBinaryChunk(response, streamState, buffer);
@@ -208,10 +204,20 @@ export function createFromReadableStream<T>(
204
options.debugChannel &&
205
options.debugChannel.readable
206
) {
211
- startReadingFromUniversalStream(response, options.debugChannel.readable);
212
- startReadingFromStream(response, stream, true);
207
+ let streamDoneCount = 0;
208
+ const handleDone = () => {
209
+ if (++streamDoneCount === 2) {
210
+ close(response);
211
+ }
212
+ };
213
+ startReadingFromUniversalStream(
214
+ response,
215
+ options.debugChannel.readable,
216
+ handleDone,
217
+ );
218
+ startReadingFromStream(response, stream, handleDone);
219
} else {
214
- startReadingFromStream(response, stream, false);
220
+ startReadingFromStream(response, stream, close.bind(null, response));
221
}
222
return getRoot(response);
223
}
@@ -250,13 +256,24 @@ export function createFromFetch<T>(
256
options.debugChannel &&
257
options.debugChannel.readable
258
) {
259
+ let streamDoneCount = 0;
260
+ const handleDone = () => {
261
+ if (++streamDoneCount === 2) {
262
+ close(response);
263
+ }
264
+ };
265
startReadingFromUniversalStream(
266
response,
267
options.debugChannel.readable,
268
+ handleDone,
269
);
257
- startReadingFromStream(response, (r.body: any), true);
270
+ startReadingFromStream(response, (r.body: any), handleDone);
271
} else {
259
- startReadingFromStream(response, (r.body: any), false);
272
+ startReadingFromStream(
273
+ response,
274
+ (r.body: any),
275
+ close.bind(null, response),
276
+ );
277
}
278
},
279
function (e) {
packages/react-server-dom-parcel/src/client/ReactFlightDOMClientEdge.js
+28
-13
@@ -102,7 +102,7 @@ function createResponseFromOptions(options?: Options) {
102
function startReadingFromStream(
103
response: FlightResponse,
104
stream: ReadableStream,
105
- isSecondaryStream: boolean,
105
+ onDone: () => void,
106
): void {
107
const streamState = createStreamState();
108
const reader = stream.getReader();
@@ -115,12 +115,7 @@ function startReadingFromStream(
115
...
116
}): void | Promise<void> {
117
if (done) {
118
- // If we're the secondary stream, then we don't close the response until
119
- // the debug channel closes.
120
- if (!isSecondaryStream) {
121
- close(response);
122
- }
123
- return;
118
+ return onDone();
119
}
120
const buffer: Uint8Array = (value: any);
121
processBinaryChunk(response, streamState, buffer);
@@ -144,10 +139,16 @@ export function createFromReadableStream<T>(
139
options.debugChannel &&
140
options.debugChannel.readable
141
) {
147
- startReadingFromStream(response, options.debugChannel.readable, false);
148
- startReadingFromStream(response, stream, true);
142
+ let streamDoneCount = 0;
143
+ const handleDone = () => {
144
+ if (++streamDoneCount === 2) {
145
+ close(response);
146
+ }
147
+ };
148
+ startReadingFromStream(response, options.debugChannel.readable, handleDone);
149
+ startReadingFromStream(response, stream, handleDone);
150
} else {
150
- startReadingFromStream(response, stream, false);
151
+ startReadingFromStream(response, stream, close.bind(null, response));
152
}
153
154
return getRoot(response);
@@ -166,10 +167,24 @@ export function createFromFetch<T>(
167
options.debugChannel &&
168
options.debugChannel.readable
169
) {
169
- startReadingFromStream(response, options.debugChannel.readable, false);
170
- startReadingFromStream(response, (r.body: any), true);
170
+ let streamDoneCount = 0;
171
+ const handleDone = () => {
172
+ if (++streamDoneCount === 2) {
173
+ close(response);
174
+ }
175
+ };
176
+ startReadingFromStream(
177
+ response,
178
+ options.debugChannel.readable,
179
+ handleDone,
180
+ );
181
+ startReadingFromStream(response, (r.body: any), handleDone);
182
} else {
172
- startReadingFromStream(response, (r.body: any), false);
183
+ startReadingFromStream(
184
+ response,
185
+ (r.body: any),
186
+ close.bind(null, response),
187
+ );
188
}
189
},
190
function (e) {
packages/react-server-dom-parcel/src/client/ReactFlightDOMClientNode.js
+11
-11
@@ -59,7 +59,7 @@ export type Options = {
59
function startReadingFromStream(
60
response: Response,
61
stream: Readable,
62
- isSecondaryStream: boolean,
62
+ onEnd: () => void,
63
): void {
64
const streamState = createStreamState();
65
@@ -75,13 +75,7 @@ function startReadingFromStream(
75
reportGlobalError(response, error);
76
});
77
78
- stream.on('end', () => {
79
- // If we're the secondary stream, then we don't close the response until the
80
- // debug channel closes.
81
- if (!isSecondaryStream) {
82
- close(response);
83
- }
84
- });
78
+ stream.on('end', onEnd);
79
}
80
81
export function createFromNodeStream<T>(
@@ -104,10 +98,16 @@ export function createFromNodeStream<T>(
98
);
99
100
if (__DEV__ && options && options.debugChannel) {
107
- startReadingFromStream(response, options.debugChannel, false);
108
- startReadingFromStream(response, stream, true);
101
+ let streamEndedCount = 0;
102
+ const handleEnd = () => {
103
+ if (++streamEndedCount === 2) {
104
+ close(response);
105
+ }
106
+ };
107
+ startReadingFromStream(response, options.debugChannel, handleEnd);
108
+ startReadingFromStream(response, stream, handleEnd);
109
} else {
110
- startReadingFromStream(response, stream, false);
110
+ startReadingFromStream(response, stream, close.bind(null, response));
111
}
112
113
return getRoot(response);
packages/react-server-dom-turbopack/src/__tests__/ReactFlightTurbopackDOMBrowser-test.js
+83
@@ -14,14 +14,20 @@ import {patchMessageChannel} from '../../../../scripts/jest/patchMessageChannel'
14
// Polyfills for test environment
15
global.ReadableStream =
16
require('web-streams-polyfill/ponyfill/es6').ReadableStream;
17
+global.WritableStream =
18
+ require('web-streams-polyfill/ponyfill/es6').WritableStream;
19
global.TextEncoder = require('util').TextEncoder;
20
global.TextDecoder = require('util').TextDecoder;
21
22
let React;
23
+let ReactDOMClient;
24
let ReactServerDOMServer;
25
let ReactServerDOMClient;
26
let ReactServerScheduler;
27
+let act;
28
let serverAct;
29
+let turbopackMap;
30
+let use;
31
32
describe('ReactFlightTurbopackDOMBrowser', () => {
33
beforeEach(() => {
@@ -36,16 +42,41 @@ describe('ReactFlightTurbopackDOMBrowser', () => {
42
jest.mock('react-server-dom-turbopack/server', () =>
43
require('react-server-dom-turbopack/server.browser'),
44
);
45
+ const TurbopackMock = require('./utils/TurbopackMock');
46
+ turbopackMap = TurbopackMock.turbopackMap;
47
48
ReactServerDOMServer = require('react-server-dom-turbopack/server.browser');
49
50
__unmockReact();
51
jest.resetModules();
52
53
+ ({act} = require('internal-test-utils'));
54
React = require('react');
55
+ ReactDOMClient = require('react-dom/client');
56
ReactServerDOMClient = require('react-server-dom-turbopack/client');
57
+ use = React.use;
58
});
59
60
+ function createDelayedStream(
61
+ stream: ReadableStream<Uint8Array>,
62
+ ): ReadableStream<Uint8Array> {
63
+ return new ReadableStream({
64
+ async start(controller) {
65
+ const reader = stream.getReader();
66
+ while (true) {
67
+ const {done, value} = await reader.read();
68
+ if (done) {
69
+ controller.close();
70
+ } else {
71
+ // Artificially delay between enqueuing chunks.
72
+ await new Promise(resolve => setTimeout(resolve));
73
+ controller.enqueue(value);
74
+ }
75
+ }
76
+ },
77
+ });
78
+ }
79
+
80
it('should resolve HTML using W3C streams', async () => {
81
function Text({children}) {
82
return <span>{children}</span>;
@@ -80,4 +111,56 @@ describe('ReactFlightTurbopackDOMBrowser', () => {
111
),
112
});
113
});
114
+
115
+ it('does not close the response early when using a fast debug channel', async () => {
116
+ function Component() {
117
+ return <div>Hi</div>;
118
+ }
119
+
120
+ let debugReadableStreamController;
121
+
122
+ const debugReadableStream = new ReadableStream({
123
+ start(controller) {
124
+ debugReadableStreamController = controller;
125
+ },
126
+ });
127
+
128
+ const rscStream = await serverAct(() =>
129
+ ReactServerDOMServer.renderToReadableStream(<Component />, turbopackMap, {
130
+ debugChannel: {
131
+ writable: new WritableStream({
132
+ write(chunk) {
133
+ debugReadableStreamController.enqueue(chunk);
134
+ },
135
+ close() {
136
+ debugReadableStreamController.close();
137
+ },
138
+ }),
139
+ },
140
+ }),
141
+ );
142
+
143
+ function ClientRoot({response}) {
144
+ return use(response);
145
+ }
146
+
147
+ const response = ReactServerDOMClient.createFromReadableStream(
148
+ // Create a delayed stream to simulate that the RSC stream might be
149
+ // transported slower than the debug channel, which must not lead to a
150
+ // `Connection closed` error in the Flight client.
151
+ createDelayedStream(rscStream),
152
+ {
153
+ debugChannel: {readable: debugReadableStream},
154
+ },
155
+ );
156
+
157
+ const container = document.createElement('div');
158
+ const root = ReactDOMClient.createRoot(container);
159
+
160
+ await act(() => {
161
+ root.render(<ClientRoot response={response} />);
162
+ });
163
+
164
+ expect(container.innerHTML).toBe('<div>Hi</div>');
165
+ });
166
});
packages/react-server-dom-turbopack/src/__tests__/ReactFlightTurbopackDOMEdge-test.js
+33
-4
@@ -78,6 +78,26 @@ describe('ReactFlightTurbopackDOMEdge', () => {
78
);
79
}
80
81
+ function createDelayedStream(
82
+ stream: ReadableStream<Uint8Array>,
83
+ ): ReadableStream<Uint8Array> {
84
+ return new ReadableStream({
85
+ async start(controller) {
86
+ const reader = stream.getReader();
87
+ while (true) {
88
+ const {done, value} = await reader.read();
89
+ if (done) {
90
+ controller.close();
91
+ } else {
92
+ // Artificially delay between enqueuing chunks.
93
+ await new Promise(resolve => setTimeout(resolve));
94
+ controller.enqueue(value);
95
+ }
96
+ }
97
+ },
98
+ });
99
+ }
100
+
101
it('should allow an alternative module mapping to be used for SSR', async () => {
102
function ClientComponent() {
103
return <span>Client Component</span>;
@@ -165,6 +185,9 @@ describe('ReactFlightTurbopackDOMEdge', () => {
185
write(chunk) {
186
debugReadableStreamController.enqueue(chunk);
187
},
188
+ close() {
189
+ debugReadableStreamController.close();
190
+ },
191
}),
192
},
193
},
@@ -184,10 +207,16 @@ describe('ReactFlightTurbopackDOMEdge', () => {
207
moduleLoading: null,
208
};
209
187
- const response = ReactServerDOMClient.createFromReadableStream(rscStream, {
188
- serverConsumerManifest,
189
- debugChannel: {readable: debugReadableStream},
190
- });
210
+ const response = ReactServerDOMClient.createFromReadableStream(
211
+ // Create a delayed stream to simulate that the RSC stream might be
212
+ // transported slower than the debug channel, which must not lead to a
213
+ // `Connection closed` error in the Flight client.
214
+ createDelayedStream(rscStream),
215
+ {
216
+ serverConsumerManifest,
217
+ debugChannel: {readable: debugReadableStream},
218
+ },
219
+ );
220
221
let ownerStack;
222
packages/react-server-dom-turbopack/src/__tests__/ReactFlightTurbopackDOMNode-test.js
+19
-1
@@ -90,6 +90,18 @@ describe('ReactFlightTurbopackDOMNode', () => {
90
);
91
}
92
93
+ function createDelayedStream() {
94
+ return new Stream.Transform({
95
+ ...streamOptions,
96
+ transform(chunk, encoding, callback) {
97
+ setTimeout(() => {
98
+ this.push(chunk);
99
+ callback();
100
+ });
101
+ },
102
+ });
103
+ }
104
+
105
it('should allow an alternative module mapping to be used for SSR', async () => {
106
function ClientComponent() {
107
return <span>Client Component</span>;
@@ -180,12 +192,18 @@ describe('ReactFlightTurbopackDOMNode', () => {
192
debugReadable.write(chunk, encoding);
193
callback();
194
},
195
+ final() {
196
+ debugReadable.end();
197
+ },
198
}),
199
},
200
),
201
);
202
188
- const readable = new Stream.PassThrough(streamOptions);
203
+ // Create a delayed stream to simulate that the RSC stream might be
204
+ // transported slower than the debug channel, which must not lead to a
205
+ // `controller.enqueueModel is not a function` error in the Flight client.
206
+ const readable = createDelayedStream();
207
208
rscStream.pipe(readable);
209
packages/react-server-dom-turbopack/src/client/ReactFlightDOMClientBrowser.js
+30
-13
@@ -100,6 +100,7 @@ function createResponseFromOptions(options: void | Options) {
100
function startReadingFromUniversalStream(
101
response: FlightResponse,
102
stream: ReadableStream,
103
+ onDone: () => void,
104
): void {
105
// This is the same as startReadingFromStream except this allows WebSocketStreams which
106
// return ArrayBuffer and string chunks instead of Uint8Array chunks. We could potentially
@@ -115,8 +116,7 @@ function startReadingFromUniversalStream(
116
...
117
}): void | Promise<void> {
118
if (done) {
118
- close(response);
119
- return;
119
+ return onDone();
120
}
121
if (value instanceof ArrayBuffer) {
122
// WebSockets can produce ArrayBuffer values in ReadableStreams.
@@ -138,7 +138,7 @@ function startReadingFromUniversalStream(
138
function startReadingFromStream(
139
response: FlightResponse,
140
stream: ReadableStream,
141
- isSecondaryStream: boolean,
141
+ onDone: () => void,
142
): void {
143
const streamState = createStreamState();
144
const reader = stream.getReader();
@@ -151,11 +151,7 @@ function startReadingFromStream(
151
...
152
}): void | Promise<void> {
153
if (done) {
154
- // If we're the secondary stream, then we don't close the response until the debug channel closes.
155
- if (!isSecondaryStream) {
156
- close(response);
157
- }
158
- return;
154
+ return onDone();
155
}
156
const buffer: Uint8Array = (value: any);
157
processBinaryChunk(response, streamState, buffer);
@@ -178,10 +174,20 @@ function createFromReadableStream<T>(
174
options.debugChannel &&
175
options.debugChannel.readable
176
) {
181
- startReadingFromUniversalStream(response, options.debugChannel.readable);
182
- startReadingFromStream(response, stream, true);
177
+ let streamDoneCount = 0;
178
+ const handleDone = () => {
179
+ if (++streamDoneCount === 2) {
180
+ close(response);
181
+ }
182
+ };
183
+ startReadingFromUniversalStream(
184
+ response,
185
+ options.debugChannel.readable,
186
+ handleDone,
187
+ );
188
+ startReadingFromStream(response, stream, handleDone);
189
} else {
184
- startReadingFromStream(response, stream, false);
190
+ startReadingFromStream(response, stream, close.bind(null, response));
191
}
192
return getRoot(response);
193
}
@@ -199,13 +205,24 @@ function createFromFetch<T>(
205
options.debugChannel &&
206
options.debugChannel.readable
207
) {
208
+ let streamDoneCount = 0;
209
+ const handleDone = () => {
210
+ if (++streamDoneCount === 2) {
211
+ close(response);
212
+ }
213
+ };
214
startReadingFromUniversalStream(
215
response,
216
options.debugChannel.readable,
217
+ handleDone,
218
);
206
- startReadingFromStream(response, (r.body: any), true);
219
+ startReadingFromStream(response, (r.body: any), handleDone);
220
} else {
208
- startReadingFromStream(response, (r.body: any), false);
221
+ startReadingFromStream(
222
+ response,
223
+ (r.body: any),
224
+ close.bind(null, response),
225
+ );
226
}
227
},
228
function (e) {
packages/react-server-dom-turbopack/src/client/ReactFlightDOMClientEdge.js
+28
-13
@@ -106,7 +106,7 @@ function createResponseFromOptions(options: Options) {
106
function startReadingFromStream(
107
response: FlightResponse,
108
stream: ReadableStream,
109
- isSecondaryStream: boolean,
109
+ onDone: () => void,
110
): void {
111
const streamState = createStreamState();
112
const reader = stream.getReader();
@@ -119,12 +119,7 @@ function startReadingFromStream(
119
...
120
}): void | Promise<void> {
121
if (done) {
122
- // If we're the secondary stream, then we don't close the response until
123
- // the debug channel closes.
124
- if (!isSecondaryStream) {
125
- close(response);
126
- }
127
- return;
122
+ return onDone();
123
}
124
const buffer: Uint8Array = (value: any);
125
processBinaryChunk(response, streamState, buffer);
@@ -148,10 +143,16 @@ function createFromReadableStream<T>(
143
options.debugChannel &&
144
options.debugChannel.readable
145
) {
151
- startReadingFromStream(response, options.debugChannel.readable, false);
152
- startReadingFromStream(response, stream, true);
146
+ let streamDoneCount = 0;
147
+ const handleDone = () => {
148
+ if (++streamDoneCount === 2) {
149
+ close(response);
150
+ }
151
+ };
152
+ startReadingFromStream(response, options.debugChannel.readable, handleDone);
153
+ startReadingFromStream(response, stream, handleDone);
154
} else {
154
- startReadingFromStream(response, stream, false);
155
+ startReadingFromStream(response, stream, close.bind(null, response));
156
}
157
158
return getRoot(response);
@@ -170,10 +171,24 @@ function createFromFetch<T>(
171
options.debugChannel &&
172
options.debugChannel.readable
173
) {
173
- startReadingFromStream(response, options.debugChannel.readable, false);
174
- startReadingFromStream(response, (r.body: any), true);
174
+ let streamDoneCount = 0;
175
+ const handleDone = () => {
176
+ if (++streamDoneCount === 2) {
177
+ close(response);
178
+ }
179
+ };
180
+ startReadingFromStream(
181
+ response,
182
+ options.debugChannel.readable,
183
+ handleDone,
184
+ );
185
+ startReadingFromStream(response, (r.body: any), handleDone);
186
} else {
176
- startReadingFromStream(response, (r.body: any), false);
187
+ startReadingFromStream(
188
+ response,
189
+ (r.body: any),
190
+ close.bind(null, response),
191
+ );
192
}
193
},
194
function (e) {
packages/react-server-dom-turbopack/src/client/ReactFlightDOMClientNode.js
+11
-11
@@ -66,7 +66,7 @@ export type Options = {
66
function startReadingFromStream(
67
response: Response,
68
stream: Readable,
69
- isSecondaryStream: boolean,
69
+ onEnd: () => void,
70
): void {
71
const streamState = createStreamState();
72
@@ -82,13 +82,7 @@ function startReadingFromStream(
82
reportGlobalError(response, error);
83
});
84
85
- stream.on('end', () => {
86
- // If we're the secondary stream, then we don't close the response until the
87
- // debug channel closes.
88
- if (!isSecondaryStream) {
89
- close(response);
90
- }
91
- });
85
+ stream.on('end', onEnd);
86
}
87
88
function createFromNodeStream<T>(
@@ -114,10 +108,16 @@ function createFromNodeStream<T>(
108
);
109
110
if (__DEV__ && options && options.debugChannel) {
117
- startReadingFromStream(response, options.debugChannel, false);
118
- startReadingFromStream(response, stream, true);
111
+ let streamEndedCount = 0;
112
+ const handleEnd = () => {
113
+ if (++streamEndedCount === 2) {
114
+ close(response);
115
+ }
116
+ };
117
+ startReadingFromStream(response, options.debugChannel, handleEnd);
118
+ startReadingFromStream(response, stream, handleEnd);
119
} else {
120
- startReadingFromStream(response, stream, false);
120
+ startReadingFromStream(response, stream, close.bind(null, response));
121
}
122
123
return getRoot(response);
packages/react-server-dom-webpack/src/__tests__/ReactFlightDOMBrowser-test.js
+74
@@ -12,6 +12,8 @@
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
@@ -152,6 +154,26 @@ describe('ReactFlightDOMBrowser', () => {
154
return fn.apply(null, args);
155
}
156
157
+ function createDelayedStream(
158
+ stream: ReadableStream<Uint8Array>,
159
+ ): ReadableStream<Uint8Array> {
160
+ return new ReadableStream({
161
+ async start(controller) {
162
+ const reader = stream.getReader();
163
+ while (true) {
164
+ const {done, value} = await reader.read();
165
+ if (done) {
166
+ controller.close();
167
+ } else {
168
+ // Artificially delay between enqueuing chunks.
169
+ await new Promise(resolve => setTimeout(resolve));
170
+ controller.enqueue(value);
171
+ }
172
+ }
173
+ },
174
+ });
175
+ }
176
+
177
it('should resolve HTML using W3C streams', async () => {
178
function Text({children}) {
179
return <span>{children}</span>;
@@ -2693,4 +2715,56 @@ describe('ReactFlightDOMBrowser', () => {
2715
2716
expect(container.innerHTML).toBe('<div></div>');
2717
});
2718
+
2719
+ it('does not close the response early when using a fast debug channel', async () => {
2720
+ function Component() {
2721
+ return <div>Hi</div>;
2722
+ }
2723
+
2724
+ let debugReadableStreamController;
2725
+
2726
+ const debugReadableStream = new ReadableStream({
2727
+ start(controller) {
2728
+ debugReadableStreamController = controller;
2729
+ },
2730
+ });
2731
+
2732
+ const rscStream = await serverAct(() =>
2733
+ ReactServerDOMServer.renderToReadableStream(<Component />, webpackMap, {
2734
+ debugChannel: {
2735
+ writable: new WritableStream({
2736
+ write(chunk) {
2737
+ debugReadableStreamController.enqueue(chunk);
2738
+ },
2739
+ close() {
2740
+ debugReadableStreamController.close();
2741
+ },
2742
+ }),
2743
+ },
2744
+ }),
2745
+ );
2746
+
2747
+ function ClientRoot({response}) {
2748
+ return use(response);
2749
+ }
2750
+
2751
+ const response = ReactServerDOMClient.createFromReadableStream(
2752
+ // Create a delayed stream to simulate that the RSC stream might be
2753
+ // transported slower than the debug channel, which must not lead to a
2754
+ // `Connection closed` error in the Flight client.
2755
+ createDelayedStream(rscStream),
2756
+ {
2757
+ debugChannel: {readable: debugReadableStream},
2758
+ },
2759
+ );
2760
+
2761
+ const container = document.createElement('div');
2762
+ const root = ReactDOMClient.createRoot(container);
2763
+
2764
+ await act(() => {
2765
+ root.render(<ClientRoot response={response} />);
2766
+ });
2767
+
2768
+ expect(container.innerHTML).toBe('<div>Hi</div>');
2769
+ });
2770
});
packages/react-server-dom-webpack/src/__tests__/ReactFlightDOMEdge-test.js
+35
-6
@@ -233,10 +233,10 @@ describe('ReactFlightDOMEdge', () => {
233
}
234
235
async function createBufferedUnclosingStream(
236
- prelude: ReadableStream<Uint8Array>,
236
+ stream: ReadableStream<Uint8Array>,
237
): ReadableStream<Uint8Array> {
238
const chunks: Array<Uint8Array> = [];
239
- const reader = prelude.getReader();
239
+ const reader = stream.getReader();
240
while (true) {
241
const {done, value} = await reader.read();
242
if (done) {
@@ -256,6 +256,26 @@ describe('ReactFlightDOMEdge', () => {
256
});
257
}
258
259
+ function createDelayedStream(
260
+ stream: ReadableStream<Uint8Array>,
261
+ ): ReadableStream<Uint8Array> {
262
+ return new ReadableStream({
263
+ async start(controller) {
264
+ const reader = stream.getReader();
265
+ while (true) {
266
+ const {done, value} = await reader.read();
267
+ if (done) {
268
+ controller.close();
269
+ } else {
270
+ // Artificially delay between enqueuing chunks.
271
+ await new Promise(resolve => setTimeout(resolve));
272
+ controller.enqueue(value);
273
+ }
274
+ }
275
+ },
276
+ });
277
+ }
278
+
279
it('should allow an alternative module mapping to be used for SSR', async () => {
280
function ClientComponent() {
281
return <span>Client Component</span>;
@@ -2012,6 +2032,9 @@ describe('ReactFlightDOMEdge', () => {
2032
write(chunk) {
2033
debugReadableStreamController.enqueue(chunk);
2034
},
2035
+ close() {
2036
+ debugReadableStreamController.close();
2037
+ },
2038
}),
2039
},
2040
},
@@ -2032,10 +2055,16 @@ describe('ReactFlightDOMEdge', () => {
2055
moduleLoading: webpackModuleLoading,
2056
};
2057
2035
- const response = ReactServerDOMClient.createFromReadableStream(rscStream, {
2036
- serverConsumerManifest,
2037
- debugChannel: {readable: debugReadableStream},
2038
- });
2058
+ const response = ReactServerDOMClient.createFromReadableStream(
2059
+ // Create a delayed stream to simulate that the RSC stream might be
2060
+ // transported slower than the debug channel, which must not lead to a
2061
+ // `Connection closed` error in the Flight client.
2062
+ createDelayedStream(rscStream),
2063
+ {
2064
+ serverConsumerManifest,
2065
+ debugChannel: {readable: debugReadableStream},
2066
+ },
2067
+ );
2068
2069
let ownerStack;
2070
packages/react-server-dom-webpack/src/__tests__/ReactFlightDOMNode-test.js
+22
-3
@@ -128,10 +128,10 @@ describe('ReactFlightDOMNode', () => {
128
}
129
130
async function createBufferedUnclosingStream(
131
- prelude: ReadableStream<Uint8Array>,
131
+ stream: ReadableStream<Uint8Array>,
132
): ReadableStream<Uint8Array> {
133
const chunks: Array<Uint8Array> = [];
134
- const reader = prelude.getReader();
134
+ const reader = stream.getReader();
135
while (true) {
136
const {done, value} = await reader.read();
137
if (done) {
@@ -151,6 +151,19 @@ describe('ReactFlightDOMNode', () => {
151
});
152
}
153
154
+ function createDelayedStream() {
155
+ return new Stream.Transform({
156
+ ...streamOptions,
157
+ transform(chunk, encoding, callback) {
158
+ // Artificially delay between pushing chunks.
159
+ setTimeout(() => {
160
+ this.push(chunk);
161
+ callback();
162
+ });
163
+ },
164
+ });
165
+ }
166
+
167
it('should support web streams in node', async () => {
168
function Text({children}) {
169
return <span>{children}</span>;
@@ -940,12 +953,18 @@ describe('ReactFlightDOMNode', () => {
953
debugReadable.write(chunk, encoding);
954
callback();
955
},
956
+ final() {
957
+ debugReadable.end();
958
+ },
959
}),
960
},
961
),
962
);
963
948
- const readable = new Stream.PassThrough(streamOptions);
964
+ // Create a delayed stream to simulate that the RSC stream might be
965
+ // transported slower than the debug channel, which must not lead to a
966
+ // `controller.enqueueModel is not a function` error in the Flight client.
967
+ const readable = createDelayedStream();
968
969
rscStream.pipe(readable);
970
packages/react-server-dom-webpack/src/client/ReactFlightDOMClientBrowser.js
+30
-13
@@ -100,6 +100,7 @@ function createResponseFromOptions(options: void | Options) {
100
function startReadingFromUniversalStream(
101
response: FlightResponse,
102
stream: ReadableStream,
103
+ onDone: () => void,
104
): void {
105
// This is the same as startReadingFromStream except this allows WebSocketStreams which
106
// return ArrayBuffer and string chunks instead of Uint8Array chunks. We could potentially
@@ -115,8 +116,7 @@ function startReadingFromUniversalStream(
116
...
117
}): void | Promise<void> {
118
if (done) {
118
- close(response);
119
- return;
119
+ return onDone();
120
}
121
if (value instanceof ArrayBuffer) {
122
// WebSockets can produce ArrayBuffer values in ReadableStreams.
@@ -138,7 +138,7 @@ function startReadingFromUniversalStream(
138
function startReadingFromStream(
139
response: FlightResponse,
140
stream: ReadableStream,
141
- isSecondaryStream: boolean,
141
+ onDone: () => void,
142
): void {
143
const streamState = createStreamState();
144
const reader = stream.getReader();
@@ -151,11 +151,7 @@ function startReadingFromStream(
151
...
152
}): void | Promise<void> {
153
if (done) {
154
- // If we're the secondary stream, then we don't close the response until the debug channel closes.
155
- if (!isSecondaryStream) {
156
- close(response);
157
- }
158
- return;
154
+ return onDone();
155
}
156
const buffer: Uint8Array = (value: any);
157
processBinaryChunk(response, streamState, buffer);
@@ -178,10 +174,20 @@ function createFromReadableStream<T>(
174
options.debugChannel &&
175
options.debugChannel.readable
176
) {
181
- startReadingFromUniversalStream(response, options.debugChannel.readable);
182
- startReadingFromStream(response, stream, true);
177
+ let streamDoneCount = 0;
178
+ const handleDone = () => {
179
+ if (++streamDoneCount === 2) {
180
+ close(response);
181
+ }
182
+ };
183
+ startReadingFromUniversalStream(
184
+ response,
185
+ options.debugChannel.readable,
186
+ handleDone,
187
+ );
188
+ startReadingFromStream(response, stream, handleDone);
189
} else {
184
- startReadingFromStream(response, stream, false);
190
+ startReadingFromStream(response, stream, close.bind(null, response));
191
}
192
return getRoot(response);
193
}
@@ -199,13 +205,24 @@ function createFromFetch<T>(
205
options.debugChannel &&
206
options.debugChannel.readable
207
) {
208
+ let streamDoneCount = 0;
209
+ const handleDone = () => {
210
+ if (++streamDoneCount === 2) {
211
+ close(response);
212
+ }
213
+ };
214
startReadingFromUniversalStream(
215
response,
216
options.debugChannel.readable,
217
+ handleDone,
218
);
206
- startReadingFromStream(response, (r.body: any), true);
219
+ startReadingFromStream(response, (r.body: any), handleDone);
220
} else {
208
- startReadingFromStream(response, (r.body: any), false);
221
+ startReadingFromStream(
222
+ response,
223
+ (r.body: any),
224
+ close.bind(null, response),
225
+ );
226
}
227
},
228
function (e) {
packages/react-server-dom-webpack/src/client/ReactFlightDOMClientEdge.js
+28
-13
@@ -106,7 +106,7 @@ function createResponseFromOptions(options: Options) {
106
function startReadingFromStream(
107
response: FlightResponse,
108
stream: ReadableStream,
109
- isSecondaryStream: boolean,
109
+ onDone: () => void,
110
): void {
111
const streamState = createStreamState();
112
const reader = stream.getReader();
@@ -119,12 +119,7 @@ function startReadingFromStream(
119
...
120
}): void | Promise<void> {
121
if (done) {
122
- // If we're the secondary stream, then we don't close the response until
123
- // the debug channel closes.
124
- if (!isSecondaryStream) {
125
- close(response);
126
- }
127
- return;
122
+ return onDone();
123
}
124
const buffer: Uint8Array = (value: any);
125
processBinaryChunk(response, streamState, buffer);
@@ -148,10 +143,16 @@ function createFromReadableStream<T>(
143
options.debugChannel &&
144
options.debugChannel.readable
145
) {
151
- startReadingFromStream(response, options.debugChannel.readable, false);
152
- startReadingFromStream(response, stream, true);
146
+ let streamDoneCount = 0;
147
+ const handleDone = () => {
148
+ if (++streamDoneCount === 2) {
149
+ close(response);
150
+ }
151
+ };
152
+ startReadingFromStream(response, options.debugChannel.readable, handleDone);
153
+ startReadingFromStream(response, stream, handleDone);
154
} else {
154
- startReadingFromStream(response, stream, false);
155
+ startReadingFromStream(response, stream, close.bind(null, response));
156
}
157
158
return getRoot(response);
@@ -170,10 +171,24 @@ function createFromFetch<T>(
171
options.debugChannel &&
172
options.debugChannel.readable
173
) {
173
- startReadingFromStream(response, options.debugChannel.readable, false);
174
- startReadingFromStream(response, (r.body: any), true);
174
+ let streamDoneCount = 0;
175
+ const handleDone = () => {
176
+ if (++streamDoneCount === 2) {
177
+ close(response);
178
+ }
179
+ };
180
+ startReadingFromStream(
181
+ response,
182
+ options.debugChannel.readable,
183
+ handleDone,
184
+ );
185
+ startReadingFromStream(response, (r.body: any), handleDone);
186
} else {
176
- startReadingFromStream(response, (r.body: any), false);
187
+ startReadingFromStream(
188
+ response,
189
+ (r.body: any),
190
+ close.bind(null, response),
191
+ );
192
}
193
},
194
function (e) {
packages/react-server-dom-webpack/src/client/ReactFlightDOMClientNode.js
+11
-11
@@ -66,7 +66,7 @@ export type Options = {
66
function startReadingFromStream(
67
response: Response,
68
stream: Readable,
69
- isSecondaryStream: boolean,
69
+ onEnd: () => void,
70
): void {
71
const streamState = createStreamState();
72
@@ -82,13 +82,7 @@ function startReadingFromStream(
82
reportGlobalError(response, error);
83
});
84
85
- stream.on('end', () => {
86
- // If we're the secondary stream, then we don't close the response until the
87
- // debug channel closes.
88
- if (!isSecondaryStream) {
89
- close(response);
90
- }
91
- });
85
+ stream.on('end', onEnd);
86
}
87
88
function createFromNodeStream<T>(
@@ -114,10 +108,16 @@ function createFromNodeStream<T>(
108
);
109
110
if (__DEV__ && options && options.debugChannel) {
117
- startReadingFromStream(response, options.debugChannel, false);
118
- startReadingFromStream(response, stream, true);
111
+ let streamEndedCount = 0;
112
+ const handleEnd = () => {
113
+ if (++streamEndedCount === 2) {
114
+ close(response);
115
+ }
116
+ };
117
+ startReadingFromStream(response, options.debugChannel, handleEnd);
118
+ startReadingFromStream(response, stream, handleEnd);
119
} else {
120
- startReadingFromStream(response, stream, false);
120
+ startReadingFromStream(response, stream, close.bind(null, response));
121
}
122
123
return getRoot(response);