[Flight] Add debugChannel option to Edge and Node clients (#34236)

When a debug channel is used between the Flight server and a browser
Flight client, we want to allow the same RSC stream to be used for
server-side rendering. To support this, the Edge and Node Flight clients
also need to accept a `debugChannel` option. Without it, debug
information would be missing (e.g. for SSR error stacks), and in some
cases this could result in `Connection closed` errors.

This PR adds support for the `debugChannel` option in the Edge and Node
clients for ESM, Parcel, Turbopack, and Webpack. Unlike the browser
clients, these clients only support a one-way channel, since the Flight
server’s return protocol is not designed for multiple clients.

The implementation follows the approach used in the browser clients, but
excludes the writable parts.

packages/react-server-dom-esm/src/client/ReactFlightDOMClientNode.js

@@ -56,8 +56,38 @@ export type Options = {
   findSourceMapURL?: FindSourceMapURLCallback,
   replayConsoleLogs?: boolean,
   environmentName?: string,
+  // For the Node.js client we only support a single-direction debug channel.
+  debugChannel?: Readable,
 };
 
+function startReadingFromStream(
+  response: Response,
+  stream: Readable,
+  isSecondaryStream: boolean,
+): void {
+  const streamState = createStreamState();
+
+  stream.on('data', chunk => {
+    if (typeof chunk === 'string') {
+      processStringChunk(response, streamState, chunk);
+    } else {
+      processBinaryChunk(response, streamState, chunk);
+    }
+  });
+
+  stream.on('error', error => {
+    reportGlobalError(response, error);
+  });
+
+  stream.on('end', () => {
+    // If we're the secondary stream, then we don't close the response until the
+    // debug channel closes.
+    if (!isSecondaryStream) {
+      close(response);
+    }
+  });
+}
+
 function createFromNodeStream<T>(
   stream: Readable,
   moduleRootPath: string,
@@ -80,18 +110,14 @@ function createFromNodeStream<T>(
       ? options.environmentName
       : undefined,
   );
-  const streamState = createStreamState();
+
-  stream.on('data', chunk => {
+  if (__DEV__ && options && options.debugChannel) {
-    if (typeof chunk === 'string') {
+    startReadingFromStream(response, options.debugChannel, false);
-      processStringChunk(response, streamState, chunk);
+    startReadingFromStream(response, stream, true);
-    } else {
+  } else {
-      processBinaryChunk(response, streamState, chunk);
+    startReadingFromStream(response, stream, false);
-    }
+  }
-  });
+
-  stream.on('error', error => {
-    reportGlobalError(response, error);
-  });
-  stream.on('end', () => close(response));
   return getRoot(response);
 }
 

packages/react-server-dom-parcel/src/client/ReactFlightDOMClientEdge.js

@@ -76,6 +76,8 @@ export type Options = {
   temporaryReferences?: TemporaryReferenceSet,
   replayConsoleLogs?: boolean,
   environmentName?: string,
+  // For the Edge client we only support a single-direction debug channel.
+  debugChannel?: {readable?: ReadableStream, ...},
 };
 
 function createResponseFromOptions(options?: Options) {
@@ -100,6 +102,7 @@ function createResponseFromOptions(options?: Options) {
 function startReadingFromStream(
   response: FlightResponse,
   stream: ReadableStream,
+  isSecondaryStream: boolean,
 ): void {
   const streamState = createStreamState();
   const reader = stream.getReader();
@@ -112,7 +115,11 @@ function startReadingFromStream(
     ...
   }): void | Promise<void> {
     if (done) {
-      close(response);
+      // If we're the secondary stream, then we don't close the response until
+      // the debug channel closes.
+      if (!isSecondaryStream) {
+        close(response);
+      }
       return;
     }
     const buffer: Uint8Array = (value: any);
@@ -130,7 +137,19 @@ export function createFromReadableStream<T>(
   options?: Options,
 ): Thenable<T> {
   const response: FlightResponse = createResponseFromOptions(options);
-  startReadingFromStream(response, stream);
+
+  if (
+    __DEV__ &&
+    options &&
+    options.debugChannel &&
+    options.debugChannel.readable
+  ) {
+    startReadingFromStream(response, options.debugChannel.readable, false);
+    startReadingFromStream(response, stream, true);
+  } else {
+    startReadingFromStream(response, stream, false);
+  }
+
   return getRoot(response);
 }
 
@@ -141,7 +160,17 @@ export function createFromFetch<T>(
   const response: FlightResponse = createResponseFromOptions(options);
   promiseForResponse.then(
     function (r) {
-      startReadingFromStream(response, (r.body: any));
+      if (
+        __DEV__ &&
+        options &&
+        options.debugChannel &&
+        options.debugChannel.readable
+      ) {
+        startReadingFromStream(response, options.debugChannel.readable, false);
+        startReadingFromStream(response, (r.body: any), true);
+      } else {
+        startReadingFromStream(response, (r.body: any), false);
+      }
     },
     function (e) {
       reportGlobalError(response, e);

packages/react-server-dom-parcel/src/client/ReactFlightDOMClientNode.js

@@ -52,8 +52,38 @@ export type Options = {
   encodeFormAction?: EncodeFormActionCallback,
   replayConsoleLogs?: boolean,
   environmentName?: string,
+  // For the Node.js client we only support a single-direction debug channel.
+  debugChannel?: Readable,
 };
 
+function startReadingFromStream(
+  response: Response,
+  stream: Readable,
+  isSecondaryStream: boolean,
+): void {
+  const streamState = createStreamState();
+
+  stream.on('data', chunk => {
+    if (typeof chunk === 'string') {
+      processStringChunk(response, streamState, chunk);
+    } else {
+      processBinaryChunk(response, streamState, chunk);
+    }
+  });
+
+  stream.on('error', error => {
+    reportGlobalError(response, error);
+  });
+
+  stream.on('end', () => {
+    // If we're the secondary stream, then we don't close the response until the
+    // debug channel closes.
+    if (!isSecondaryStream) {
+      close(response);
+    }
+  });
+}
+
 export function createFromNodeStream<T>(
   stream: Readable,
   options?: Options,
@@ -72,17 +102,13 @@ export function createFromNodeStream<T>(
       ? options.environmentName
       : undefined,
   );
-  const streamState = createStreamState();
+
-  stream.on('data', chunk => {
+  if (__DEV__ && options && options.debugChannel) {
-    if (typeof chunk === 'string') {
+    startReadingFromStream(response, options.debugChannel, false);
-      processStringChunk(response, streamState, chunk);
+    startReadingFromStream(response, stream, true);
-    } else {
+  } else {
-      processBinaryChunk(response, streamState, chunk);
+    startReadingFromStream(response, stream, false);
-    }
+  }
-  });
+
-  stream.on('error', error => {
-    reportGlobalError(response, error);
-  });
-  stream.on('end', () => close(response));
   return getRoot(response);
 }

packages/react-server-dom-turbopack/src/__tests__/ReactFlightTurbopackDOMEdge-test.js

@@ -12,26 +12,28 @@
 // Polyfills for test environment
 global.ReadableStream =
   require('web-streams-polyfill/ponyfill/es6').ReadableStream;
+global.WritableStream =
+  require('web-streams-polyfill/ponyfill/es6').WritableStream;
 global.TextEncoder = require('util').TextEncoder;
 global.TextDecoder = require('util').TextDecoder;
 
-// Don't wait before processing work on the server.
-// TODO: we can replace this with FlightServer.act().
-global.setTimeout = cb => cb();
-
 let clientExports;
 let turbopackMap;
 let turbopackModules;
 let React;
+let ReactServer;
 let ReactDOMServer;
 let ReactServerDOMServer;
 let ReactServerDOMClient;
 let use;
+let serverAct;
 
 describe('ReactFlightTurbopackDOMEdge', () => {
   beforeEach(() => {
     jest.resetModules();
 
+    serverAct = require('internal-test-utils').serverAct;
+
     // Simulate the condition resolution
     jest.mock('react', () => require('react/react.react-server'));
     jest.mock('react-server-dom-turbopack/server', () =>
@@ -43,6 +45,7 @@ describe('ReactFlightTurbopackDOMEdge', () => {
     turbopackMap = TurbopackMock.turbopackMap;
     turbopackModules = TurbopackMock.turbopackModules;
 
+    ReactServer = require('react');
     ReactServerDOMServer = require('react-server-dom-turbopack/server.edge');
 
     jest.resetModules();
@@ -66,6 +69,15 @@ describe('ReactFlightTurbopackDOMEdge', () => {
     }
   }
 
+  function normalizeCodeLocInfo(str) {
+    return (
+      str &&
+      str.replace(/^ +(?:at|in) ([\S]+)[^\n]*/gm, function (m, name) {
+        return '    in ' + name + (/\d/.test(m) ? ' (at **)' : '');
+      })
+    );
+  }
+
   it('should allow an alternative module mapping to be used for SSR', async () => {
     function ClientComponent() {
       return <span>Client Component</span>;
@@ -92,9 +104,8 @@ describe('ReactFlightTurbopackDOMEdge', () => {
       return <ClientComponentOnTheClient />;
     }
 
-    const stream = ReactServerDOMServer.renderToReadableStream(
+    const stream = await serverAct(() =>
-      <App />,
+      ReactServerDOMServer.renderToReadableStream(<App />, turbopackMap),
-      turbopackMap,
     );
     const response = ReactServerDOMClient.createFromReadableStream(stream, {
       serverConsumerManifest: {
@@ -107,10 +118,98 @@ describe('ReactFlightTurbopackDOMEdge', () => {
       return use(response);
     }
 
-    const ssrStream = await ReactDOMServer.renderToReadableStream(
+    const ssrStream = await serverAct(() =>
-      <ClientRoot />,
+      ReactDOMServer.renderToReadableStream(<ClientRoot />),
     );
     const result = await readResult(ssrStream);
     expect(result).toEqual('<span>Client Component</span>');
   });
+
+  // @gate __DEV__
+  it('can transport debug info through a separate debug channel', async () => {
+    function Thrower() {
+      throw new Error('ssr-throw');
+    }
+
+    const ClientComponentOnTheClient = clientExports(
+      Thrower,
+      123,
+      'path/to/chunk.js',
+    );
+
+    const ClientComponentOnTheServer = clientExports(Thrower);
+
+    function App() {
+      return ReactServer.createElement(
+        ReactServer.Suspense,
+        null,
+        ReactServer.createElement(ClientComponentOnTheClient, null),
+      );
+    }
+
+    let debugReadableStreamController;
+
+    const debugReadableStream = new ReadableStream({
+      start(controller) {
+        debugReadableStreamController = controller;
+      },
+    });
+
+    const rscStream = await serverAct(() =>
+      ReactServerDOMServer.renderToReadableStream(
+        ReactServer.createElement(App, null),
+        turbopackMap,
+        {
+          debugChannel: {
+            writable: new WritableStream({
+              write(chunk) {
+                debugReadableStreamController.enqueue(chunk);
+              },
+            }),
+          },
+        },
+      ),
+    );
+
+    function ClientRoot({response}) {
+      return use(response);
+    }
+
+    const serverConsumerManifest = {
+      moduleMap: {
+        [turbopackMap[ClientComponentOnTheClient.$$id].id]: {
+          '*': turbopackMap[ClientComponentOnTheServer.$$id],
+        },
+      },
+      moduleLoading: null,
+    };
+
+    const response = ReactServerDOMClient.createFromReadableStream(rscStream, {
+      serverConsumerManifest,
+      debugChannel: {readable: debugReadableStream},
+    });
+
+    let ownerStack;
+
+    const ssrStream = await serverAct(() =>
+      ReactDOMServer.renderToReadableStream(
+        <ClientRoot response={response} />,
+        {
+          onError(err, errorInfo) {
+            ownerStack = React.captureOwnerStack
+              ? React.captureOwnerStack()
+              : null;
+          },
+        },
+      ),
+    );
+
+    const result = await readResult(ssrStream);
+
+    expect(normalizeCodeLocInfo(ownerStack)).toBe('\n    in App (at **)');
+
+    expect(result).toContain(
+      'Switched to client rendering because the server rendering errored:\n\nssr-throw',
+    );
+  });
 });

packages/react-server-dom-turbopack/src/__tests__/ReactFlightTurbopackDOMNode-test.js

@@ -17,12 +17,17 @@ let turbopackModules;
 let turbopackModuleLoading;
 let React;
 let ReactDOMServer;
+let ReactServer;
 let ReactServerDOMServer;
 let ReactServerDOMClient;
 let Stream;
 let use;
 let serverAct;
 
+const streamOptions = {
+  objectMode: true,
+};
+
 describe('ReactFlightTurbopackDOMNode', () => {
   beforeEach(() => {
     jest.resetModules();
@@ -35,6 +40,7 @@ describe('ReactFlightTurbopackDOMNode', () => {
     jest.mock('react-server-dom-turbopack/server', () =>
       require('react-server-dom-turbopack/server.node'),
     );
+    ReactServer = require('react');
     ReactServerDOMServer = require('react-server-dom-turbopack/server');
 
     const TurbopackMock = require('./utils/TurbopackMock');
@@ -75,6 +81,15 @@ describe('ReactFlightTurbopackDOMNode', () => {
     });
   }
 
+  function normalizeCodeLocInfo(str) {
+    return (
+      str &&
+      str.replace(/^ +(?:at|in) ([\S]+)[^\n]*/gm, function (m, name) {
+        return '    in ' + name + (/\d/.test(m) ? ' (at **)' : '');
+      })
+    );
+  }
+
   it('should allow an alternative module mapping to be used for SSR', async () => {
     function ClientComponent() {
       return <span>Client Component</span>;
@@ -130,4 +145,90 @@ describe('ReactFlightTurbopackDOMNode', () => {
       '<script src="/prefix/path/to/chunk.js" async=""></script><span>Client Component</span>',
     );
   });
+
+  // @gate __DEV__
+  it('can transport debug info through a separate debug channel', async () => {
+    function Thrower() {
+      throw new Error('ssr-throw');
+    }
+
+    const ClientComponentOnTheClient = clientExports(
+      Thrower,
+      123,
+      'path/to/chunk.js',
+    );
+
+    const ClientComponentOnTheServer = clientExports(Thrower);
+
+    function App() {
+      return ReactServer.createElement(
+        ReactServer.Suspense,
+        null,
+        ReactServer.createElement(ClientComponentOnTheClient, null),
+      );
+    }
+
+    const debugReadable = new Stream.PassThrough(streamOptions);
+
+    const rscStream = await serverAct(() =>
+      ReactServerDOMServer.renderToPipeableStream(
+        ReactServer.createElement(App, null),
+        turbopackMap,
+        {
+          debugChannel: new Stream.Writable({
+            write(chunk, encoding, callback) {
+              debugReadable.write(chunk, encoding);
+              callback();
+            },
+          }),
+        },
+      ),
+    );
+
+    const readable = new Stream.PassThrough(streamOptions);
+
+    rscStream.pipe(readable);
+
+    function ClientRoot({response}) {
+      return use(response);
+    }
+
+    const serverConsumerManifest = {
+      moduleMap: {
+        [turbopackMap[ClientComponentOnTheClient.$$id].id]: {
+          '*': turbopackMap[ClientComponentOnTheServer.$$id],
+        },
+      },
+      moduleLoading: null,
+    };
+
+    const response = ReactServerDOMClient.createFromNodeStream(
+      readable,
+      serverConsumerManifest,
+      {debugChannel: debugReadable},
+    );
+
+    let ownerStack;
+
+    const ssrStream = await serverAct(() =>
+      ReactDOMServer.renderToPipeableStream(
+        <ClientRoot response={response} />,
+        {
+          onError(err, errorInfo) {
+            ownerStack = React.captureOwnerStack
+              ? React.captureOwnerStack()
+              : null;
+          },
+        },
+      ),
+    );
+
+    const result = await readResult(ssrStream);
+
+    expect(normalizeCodeLocInfo(ownerStack)).toBe('\n    in App (at **)');
+
+    expect(result).toContain(
+      'Switched to client rendering because the server rendering errored:\n\nssr-throw',
+    );
+  });
 });

packages/react-server-dom-turbopack/src/client/ReactFlightDOMClientEdge.js

@@ -78,6 +78,8 @@ export type Options = {
   findSourceMapURL?: FindSourceMapURLCallback,
   replayConsoleLogs?: boolean,
   environmentName?: string,
+  // For the Edge client we only support a single-direction debug channel.
+  debugChannel?: {readable?: ReadableStream, ...},
 };
 
 function createResponseFromOptions(options: Options) {
@@ -104,6 +106,7 @@ function createResponseFromOptions(options: Options) {
 function startReadingFromStream(
   response: FlightResponse,
   stream: ReadableStream,
+  isSecondaryStream: boolean,
 ): void {
   const streamState = createStreamState();
   const reader = stream.getReader();
@@ -116,7 +119,11 @@ function startReadingFromStream(
     ...
   }): void | Promise<void> {
     if (done) {
-      close(response);
+      // If we're the secondary stream, then we don't close the response until
+      // the debug channel closes.
+      if (!isSecondaryStream) {
+        close(response);
+      }
       return;
     }
     const buffer: Uint8Array = (value: any);
@@ -134,7 +141,19 @@ function createFromReadableStream<T>(
   options: Options,
 ): Thenable<T> {
   const response: FlightResponse = createResponseFromOptions(options);
-  startReadingFromStream(response, stream);
+
+  if (
+    __DEV__ &&
+    options &&
+    options.debugChannel &&
+    options.debugChannel.readable
+  ) {
+    startReadingFromStream(response, options.debugChannel.readable, false);
+    startReadingFromStream(response, stream, true);
+  } else {
+    startReadingFromStream(response, stream, false);
+  }
+
   return getRoot(response);
 }
 
@@ -145,7 +164,17 @@ function createFromFetch<T>(
   const response: FlightResponse = createResponseFromOptions(options);
   promiseForResponse.then(
     function (r) {
-      startReadingFromStream(response, (r.body: any));
+      if (
+        __DEV__ &&
+        options &&
+        options.debugChannel &&
+        options.debugChannel.readable
+      ) {
+        startReadingFromStream(response, options.debugChannel.readable, false);
+        startReadingFromStream(response, (r.body: any), true);
+      } else {
+        startReadingFromStream(response, (r.body: any), false);
+      }
     },
     function (e) {
       reportGlobalError(response, e);

packages/react-server-dom-turbopack/src/client/ReactFlightDOMClientNode.js

@@ -59,8 +59,38 @@ export type Options = {
   findSourceMapURL?: FindSourceMapURLCallback,
   replayConsoleLogs?: boolean,
   environmentName?: string,
+  // For the Node.js client we only support a single-direction debug channel.
+  debugChannel?: Readable,
 };
 
+function startReadingFromStream(
+  response: Response,
+  stream: Readable,
+  isSecondaryStream: boolean,
+): void {
+  const streamState = createStreamState();
+
+  stream.on('data', chunk => {
+    if (typeof chunk === 'string') {
+      processStringChunk(response, streamState, chunk);
+    } else {
+      processBinaryChunk(response, streamState, chunk);
+    }
+  });
+
+  stream.on('error', error => {
+    reportGlobalError(response, error);
+  });
+
+  stream.on('end', () => {
+    // If we're the secondary stream, then we don't close the response until the
+    // debug channel closes.
+    if (!isSecondaryStream) {
+      close(response);
+    }
+  });
+}
+
 function createFromNodeStream<T>(
   stream: Readable,
   serverConsumerManifest: ServerConsumerManifest,
@@ -82,18 +112,14 @@ function createFromNodeStream<T>(
       ? options.environmentName
       : undefined,
   );
-  const streamState = createStreamState();
+
-  stream.on('data', chunk => {
+  if (__DEV__ && options && options.debugChannel) {
-    if (typeof chunk === 'string') {
+    startReadingFromStream(response, options.debugChannel, false);
-      processStringChunk(response, streamState, chunk);
+    startReadingFromStream(response, stream, true);
-    } else {
+  } else {
-      processBinaryChunk(response, streamState, chunk);
+    startReadingFromStream(response, stream, false);
-    }
+  }
-  });
+
-  stream.on('error', error => {
-    reportGlobalError(response, error);
-  });
-  stream.on('end', () => close(response));
   return getRoot(response);
 }
 

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

@@ -13,6 +13,8 @@
 // Polyfills for test environment
 global.ReadableStream =
   require('web-streams-polyfill/ponyfill/es6').ReadableStream;
+global.WritableStream =
+  require('web-streams-polyfill/ponyfill/es6').WritableStream;
 global.TextEncoder = require('util').TextEncoder;
 global.TextDecoder = require('util').TextDecoder;
 global.Blob = require('buffer').Blob;
@@ -1968,4 +1970,94 @@ describe('ReactFlightDOMEdge', () => {
     const result = await readResult(ssrStream);
     expect(result).toEqual('<div></div>');
   });
+
+  // @gate __DEV__
+  it('can transport debug info through a separate debug channel', async () => {
+    function Thrower() {
+      throw new Error('ssr-throw');
+    }
+
+    const ClientComponentOnTheClient = clientExports(
+      Thrower,
+      123,
+      'path/to/chunk.js',
+    );
+
+    const ClientComponentOnTheServer = clientExports(Thrower);
+
+    function App() {
+      return ReactServer.createElement(
+        ReactServer.Suspense,
+        null,
+        ReactServer.createElement(ClientComponentOnTheClient, null),
+      );
+    }
+
+    let debugReadableStreamController;
+
+    const debugReadableStream = new ReadableStream({
+      start(controller) {
+        debugReadableStreamController = controller;
+      },
+    });
+
+    const rscStream = await serverAct(() =>
+      passThrough(
+        ReactServerDOMServer.renderToReadableStream(
+          ReactServer.createElement(App, null),
+          webpackMap,
+          {
+            debugChannel: {
+              writable: new WritableStream({
+                write(chunk) {
+                  debugReadableStreamController.enqueue(chunk);
+                },
+              }),
+            },
+          },
+        ),
+      ),
+    );
+
+    function ClientRoot({response}) {
+      return use(response);
+    }
+
+    const serverConsumerManifest = {
+      moduleMap: {
+        [webpackMap[ClientComponentOnTheClient.$$id].id]: {
+          '*': webpackMap[ClientComponentOnTheServer.$$id],
+        },
+      },
+      moduleLoading: webpackModuleLoading,
+    };
+
+    const response = ReactServerDOMClient.createFromReadableStream(rscStream, {
+      serverConsumerManifest,
+      debugChannel: {readable: debugReadableStream},
+    });
+
+    let ownerStack;
+
+    const ssrStream = await serverAct(() =>
+      ReactDOMServer.renderToReadableStream(
+        <ClientRoot response={response} />,
+        {
+          onError(err, errorInfo) {
+            ownerStack = React.captureOwnerStack
+              ? React.captureOwnerStack()
+              : null;
+          },
+        },
+      ),
+    );
+
+    const result = await readResult(ssrStream);
+
+    expect(normalizeCodeLocInfo(ownerStack)).toBe('\n    in App (at **)');
+
+    expect(result).toContain(
+      'Switched to client rendering because the server rendering errored:\n\nssr-throw',
+    );
+  });
 });

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

@@ -905,4 +905,90 @@ describe('ReactFlightDOMNode', () => {
     // We don't really have an assertion other than to make sure
     // the stream doesn't hang.
   });
+
+  // @gate __DEV__
+  it('can transport debug info through a separate debug channel', async () => {
+    function Thrower() {
+      throw new Error('ssr-throw');
+    }
+
+    const ClientComponentOnTheClient = clientExports(
+      Thrower,
+      123,
+      'path/to/chunk.js',
+    );
+
+    const ClientComponentOnTheServer = clientExports(Thrower);
+
+    function App() {
+      return ReactServer.createElement(
+        ReactServer.Suspense,
+        null,
+        ReactServer.createElement(ClientComponentOnTheClient, null),
+      );
+    }
+
+    const debugReadable = new Stream.PassThrough(streamOptions);
+
+    const rscStream = await serverAct(() =>
+      ReactServerDOMServer.renderToPipeableStream(
+        ReactServer.createElement(App, null),
+        webpackMap,
+        {
+          debugChannel: new Stream.Writable({
+            write(chunk, encoding, callback) {
+              debugReadable.write(chunk, encoding);
+              callback();
+            },
+          }),
+        },
+      ),
+    );
+
+    const readable = new Stream.PassThrough(streamOptions);
+
+    rscStream.pipe(readable);
+
+    function ClientRoot({response}) {
+      return use(response);
+    }
+
+    const serverConsumerManifest = {
+      moduleMap: {
+        [webpackMap[ClientComponentOnTheClient.$$id].id]: {
+          '*': webpackMap[ClientComponentOnTheServer.$$id],
+        },
+      },
+      moduleLoading: webpackModuleLoading,
+    };
+
+    const response = ReactServerDOMClient.createFromNodeStream(
+      readable,
+      serverConsumerManifest,
+      {debugChannel: debugReadable},
+    );
+
+    let ownerStack;
+
+    const ssrStream = await serverAct(() =>
+      ReactDOMServer.renderToPipeableStream(
+        <ClientRoot response={response} />,
+        {
+          onError(err, errorInfo) {
+            ownerStack = React.captureOwnerStack
+              ? React.captureOwnerStack()
+              : null;
+          },
+        },
+      ),
+    );
+
+    const result = await readResult(ssrStream);
+
+    expect(normalizeCodeLocInfo(ownerStack)).toBe('\n    in App (at **)');
+
+    expect(result).toContain(
+      'Switched to client rendering because the server rendering errored:\n\nssr-throw',
+    );
+  });
 });

packages/react-server-dom-webpack/src/client/ReactFlightDOMClientEdge.js

@@ -78,6 +78,8 @@ export type Options = {
   findSourceMapURL?: FindSourceMapURLCallback,
   replayConsoleLogs?: boolean,
   environmentName?: string,
+  // For the Edge client we only support a single-direction debug channel.
+  debugChannel?: {readable?: ReadableStream, ...},
 };
 
 function createResponseFromOptions(options: Options) {
@@ -104,6 +106,7 @@ function createResponseFromOptions(options: Options) {
 function startReadingFromStream(
   response: FlightResponse,
   stream: ReadableStream,
+  isSecondaryStream: boolean,
 ): void {
   const streamState = createStreamState();
   const reader = stream.getReader();
@@ -116,7 +119,11 @@ function startReadingFromStream(
     ...
   }): void | Promise<void> {
     if (done) {
-      close(response);
+      // If we're the secondary stream, then we don't close the response until
+      // the debug channel closes.
+      if (!isSecondaryStream) {
+        close(response);
+      }
       return;
     }
     const buffer: Uint8Array = (value: any);
@@ -134,7 +141,19 @@ function createFromReadableStream<T>(
   options: Options,
 ): Thenable<T> {
   const response: FlightResponse = createResponseFromOptions(options);
-  startReadingFromStream(response, stream);
+
+  if (
+    __DEV__ &&
+    options &&
+    options.debugChannel &&
+    options.debugChannel.readable
+  ) {
+    startReadingFromStream(response, options.debugChannel.readable, false);
+    startReadingFromStream(response, stream, true);
+  } else {
+    startReadingFromStream(response, stream, false);
+  }
+
   return getRoot(response);
 }
 
@@ -145,7 +164,17 @@ function createFromFetch<T>(
   const response: FlightResponse = createResponseFromOptions(options);
   promiseForResponse.then(
     function (r) {
-      startReadingFromStream(response, (r.body: any));
+      if (
+        __DEV__ &&
+        options &&
+        options.debugChannel &&
+        options.debugChannel.readable
+      ) {
+        startReadingFromStream(response, options.debugChannel.readable, false);
+        startReadingFromStream(response, (r.body: any), true);
+      } else {
+        startReadingFromStream(response, (r.body: any), false);
+      }
     },
     function (e) {
       reportGlobalError(response, e);

packages/react-server-dom-webpack/src/client/ReactFlightDOMClientNode.js

@@ -59,8 +59,38 @@ export type Options = {
   findSourceMapURL?: FindSourceMapURLCallback,
   replayConsoleLogs?: boolean,
   environmentName?: string,
+  // For the Node.js client we only support a single-direction debug channel.
+  debugChannel?: Readable,
 };
 
+function startReadingFromStream(
+  response: Response,
+  stream: Readable,
+  isSecondaryStream: boolean,
+): void {
+  const streamState = createStreamState();
+
+  stream.on('data', chunk => {
+    if (typeof chunk === 'string') {
+      processStringChunk(response, streamState, chunk);
+    } else {
+      processBinaryChunk(response, streamState, chunk);
+    }
+  });
+
+  stream.on('error', error => {
+    reportGlobalError(response, error);
+  });
+
+  stream.on('end', () => {
+    // If we're the secondary stream, then we don't close the response until the
+    // debug channel closes.
+    if (!isSecondaryStream) {
+      close(response);
+    }
+  });
+}
+
 function createFromNodeStream<T>(
   stream: Readable,
   serverConsumerManifest: ServerConsumerManifest,
@@ -82,18 +112,14 @@ function createFromNodeStream<T>(
       ? options.environmentName
       : undefined,
   );
-  const streamState = createStreamState();
+
-  stream.on('data', chunk => {
+  if (__DEV__ && options && options.debugChannel) {
-    if (typeof chunk === 'string') {
+    startReadingFromStream(response, options.debugChannel, false);
-      processStringChunk(response, streamState, chunk);
+    startReadingFromStream(response, stream, true);
-    } else {
+  } else {
-      processBinaryChunk(response, streamState, chunk);
+    startReadingFromStream(response, stream, false);
-    }
+  }
-  });
+
-  stream.on('error', error => {
-    reportGlobalError(response, error);
-  });
-  stream.on('end', () => close(response));
   return getRoot(response);
 }