[Flight] Use more robust web socket implementation in fixture (#34338)
The `WebSocketStream` implementation seems to be a bit unreliable. We've seen `Cannot close a ERRORED writable stream` errors when expanding the logged deep object, for example. And when reducing the fixture to a minimal app, we even get `Connection closed` errors, because the web socket connection is closed before all debug chunks are sent. We can improve the reliability of the web socket connection by using a normal `WebSocket` instance on the client, along with manually creating a `WritableStream` and a `ReadableStream` for processing the messages. As an additional benefit, the debug channel now also works in Firefox and Safari. On the server, we're simplifying the integration with the Express server a bit by utilizing the `server` property for `WebSocket.Server`, instead of the `noServer` property with the manual upgrade handling.
Hendrik Liebau committed
Aug 29, 2025 at 12:04 UTC
3fe51c9e147fb5215011cf3fd9544c5a776abf41
2 files changed
+61
-40
fixtures/flight/server/region.js
+17
-26
@@ -74,13 +74,7 @@ function getDebugChannel(req) {
74
return activeDebugChannels.get(requestId);
75
}
76
77
-async function renderApp(
78
- res,
79
- returnValue,
80
- formState,
81
- noCache,
82
- promiseForDebugChannel
83
-) {
77
+async function renderApp(res, returnValue, formState, noCache, debugChannel) {
78
const {renderToPipeableStream} = await import(
79
'react-server-dom-webpack/server'
80
);
@@ -132,7 +126,7 @@ async function renderApp(
126
// For client-invoked server actions we refresh the tree and return a return value.
127
const payload = {root, returnValue, formState};
128
const {pipe} = renderToPipeableStream(payload, moduleMap, {
135
- debugChannel: await promiseForDebugChannel,
129
+ debugChannel,
130
filterStackFrame,
131
});
132
pipe(res);
@@ -385,23 +379,20 @@ app.on('error', function (error) {
379
if (process.env.NODE_ENV === 'development') {
380
// Open a websocket server for Debug information
381
const WebSocket = require('ws');
388
- const webSocketServer = new WebSocket.Server({noServer: true});
389
-
390
- httpServer.on('upgrade', (request, socket, head) => {
391
- const DEBUG_CHANNEL_PATH = '/debug-channel?';
392
- if (request.url.startsWith(DEBUG_CHANNEL_PATH)) {
393
- const requestId = request.url.slice(DEBUG_CHANNEL_PATH.length);
394
- const promiseForWs = new Promise(resolve => {
395
- webSocketServer.handleUpgrade(request, socket, head, ws => {
396
- ws.on('close', () => {
397
- activeDebugChannels.delete(requestId);
398
- });
399
- resolve(ws);
400
- });
401
- });
402
- activeDebugChannels.set(requestId, promiseForWs);
403
- } else {
404
- socket.destroy();
405
- }
382
+
383
+ const webSocketServer = new WebSocket.Server({
384
+ server: httpServer,
385
+ path: '/debug-channel',
386
+ });
387
+
388
+ webSocketServer.on('connection', (ws, req) => {
389
+ const url = new URL(req.url, `http://${req.headers.host}`);
390
+ const requestId = url.searchParams.get('id');
391
+
392
+ activeDebugChannels.set(requestId, ws);
393
+
394
+ ws.on('close', (code, reason) => {
395
+ activeDebugChannels.delete(requestId);
396
+ });
397
});
398
}
fixtures/flight/src/index.js
+44
-14
@@ -14,18 +14,52 @@ function findSourceMapURL(fileName) {
14
);
15
}
16
17
+async function createWebSocketStream(url) {
18
+ const ws = new WebSocket(url);
19
+ ws.binaryType = 'arraybuffer';
20
+
21
+ await new Promise((resolve, reject) => {
22
+ ws.addEventListener('open', resolve, {once: true});
23
+ ws.addEventListener('error', reject, {once: true});
24
+ });
25
+
26
+ const writable = new WritableStream({
27
+ write(chunk) {
28
+ ws.send(chunk);
29
+ },
30
+ close() {
31
+ ws.close();
32
+ },
33
+ abort(reason) {
34
+ ws.close(1000, reason && String(reason));
35
+ },
36
+ });
37
+
38
+ const readable = new ReadableStream({
39
+ start(controller) {
40
+ ws.addEventListener('message', event => {
41
+ controller.enqueue(event.data);
42
+ });
43
+ ws.addEventListener('close', () => {
44
+ controller.close();
45
+ });
46
+ ws.addEventListener('error', err => {
47
+ controller.error(err);
48
+ });
49
+ },
50
+ });
51
+
52
+ return {readable, writable};
53
+}
54
+
55
let updateRoot;
56
async function callServer(id, args) {
57
let response;
20
- if (
21
- process.env.NODE_ENV === 'development' &&
22
- typeof WebSocketStream === 'function'
23
- ) {
58
+ if (process.env.NODE_ENV === 'development') {
59
const requestId = crypto.randomUUID();
25
- const wss = new WebSocketStream(
26
- 'ws://localhost:3001/debug-channel?' + requestId
60
+ const debugChannel = await createWebSocketStream(
61
+ `ws://localhost:3001/debug-channel?id=${requestId}`
62
);
28
- const debugChannel = await wss.opened;
63
response = createFromFetch(
64
fetch('/', {
65
method: 'POST',
@@ -74,15 +108,11 @@ function Shell({data}) {
108
109
async function hydrateApp() {
110
let response;
77
- if (
78
- process.env.NODE_ENV === 'development' &&
79
- typeof WebSocketStream === 'function'
80
- ) {
111
+ if (process.env.NODE_ENV === 'development') {
112
const requestId = crypto.randomUUID();
82
- const wss = new WebSocketStream(
83
- 'ws://localhost:3001/debug-channel?' + requestId
113
+ const debugChannel = await createWebSocketStream(
114
+ `ws://localhost:3001/debug-channel?id=${requestId}`
115
);
85
- const debugChannel = await wss.opened;
116
response = createFromFetch(
117
fetch('/', {
118
headers: {