[Flight] Send server reference error chunks to the client (#26293)

Previously when a called server reference function was rejected, the
emitted error chunk was not flushed, and the request was not properly
closed.

Co-authored-by: Sebastian Markbage <sebastian@calyptus.eu>

fixtures/flight/src/Button.js

@@ -3,11 +3,21 @@
 import * as React from 'react';
 
 export default function Button({action, children}) {
+  const [isPending, setIsPending] = React.useState(false);
+
   return (
     <button
+      disabled={isPending}
       onClick={async () => {
-        const result = await action();
-        console.log(result);
+        setIsPending(true);
+        try {
+          const result = await action();
+          console.log(result);
+        } catch (error) {
+          console.error(error);
+        } finally {
+          setIsPending(false);
+        }
       }}>
       {children}
     </button>

fixtures/flight/src/actions.js

@@ -1,6 +1,13 @@
 'use server';
 
 export async function like() {
-  console.log('Like');
-  return 'Liked';
+  return new Promise((resolve, reject) =>
+    setTimeout(
+      () =>
+        Math.random() > 0.5
+          ? resolve('Liked')
+          : reject(new Error('Failed to like')),
+      500
+    )
+  );
 }

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

@@ -1046,4 +1046,72 @@ describe('ReactFlightDOM', () => {
     });
     expect(container.innerHTML).toBe('<p>async hello</p>');
   });
+
+  // @gate enableUseHook
+  it('should throw on the client if a passed promise eventually rejects', async () => {
+    const reportedErrors = [];
+    const theError = new Error('Server throw');
+
+    async function getData() {
+      throw theError;
+    }
+
+    function Component({data}) {
+      const text = use(data);
+      return <p>{text}</p>;
+    }
+
+    const ClientComponent = clientExports(Component);
+
+    function ServerComponent() {
+      const data = getData(); // no await here
+      return <ClientComponent data={data} />;
+    }
+
+    function Await({response}) {
+      return use(response);
+    }
+
+    function App({response}) {
+      return (
+        <Suspense fallback={<h1>Loading...</h1>}>
+          <ErrorBoundary
+            fallback={e => (
+              <p>
+                {__DEV__ ? e.message + ' + ' : null}
+                {e.digest}
+              </p>
+            )}>
+            <Await response={response} />
+          </ErrorBoundary>
+        </Suspense>
+      );
+    }
+
+    const {writable, readable} = getTestStream();
+    const {pipe} = ReactServerDOMWriter.renderToPipeableStream(
+      <ServerComponent />,
+      webpackMap,
+      {
+        onError(x) {
+          reportedErrors.push(x);
+          return __DEV__ ? 'a dev digest' : `digest("${x.message}")`;
+        },
+      },
+    );
+    pipe(writable);
+    const response = ReactServerDOMReader.createFromReadableStream(readable);
+
+    const container = document.createElement('div');
+    const root = ReactDOMClient.createRoot(container);
+    await act(async () => {
+      root.render(<App response={response} />);
+    });
+    expect(container.innerHTML).toBe(
+      __DEV__
+        ? '<p>Server throw + a dev digest</p>'
+        : '<p>digest("Server throw")</p>',
+    );
+    expect(reportedErrors).toEqual([theError]);
+  });
 });

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

@@ -850,4 +850,68 @@ describe('ReactFlightDOMBrowser', () => {
     const result = await actionProxy('!');
     expect(result).toBe('Hello World!');
   });
+
+  it('propagates server reference errors to the client', async () => {
+    let actionProxy;
+
+    function Client({action}) {
+      actionProxy = action;
+      return 'Click Me';
+    }
+
+    async function send(text) {
+      return Promise.reject(new Error(`Error for ${text}`));
+    }
+
+    const ServerModule = serverExports({send});
+    const ClientRef = clientExports(Client);
+
+    const stream = ReactServerDOMWriter.renderToReadableStream(
+      <ClientRef action={ServerModule.send} />,
+      webpackMap,
+    );
+
+    const response = ReactServerDOMReader.createFromReadableStream(stream, {
+      async callServer(ref, args) {
+        const fn = requireServerRef(ref);
+        return ReactServerDOMReader.createFromReadableStream(
+          ReactServerDOMWriter.renderToReadableStream(
+            fn.apply(null, args),
+            null,
+            {onError: error => 'test-error-digest'},
+          ),
+        );
+      },
+    });
+
+    function App() {
+      return use(response);
+    }
+
+    const container = document.createElement('div');
+    const root = ReactDOMClient.createRoot(container);
+    await act(async () => {
+      root.render(<App />);
+    });
+
+    if (__DEV__) {
+      await expect(actionProxy('test')).rejects.toThrow('Error for test');
+    } else {
+      let thrownError;
+
+      try {
+        await actionProxy('test');
+      } catch (error) {
+        thrownError = error;
+      }
+
+      expect(thrownError).toEqual(
+        new Error(
+          'An error occurred in the Server Components render. The specific message is omitted in production builds to avoid leaking sensitive details. A digest property is included on this error instance which may provide additional details about the nature of the error.',
+        ),
+      );
+
+      expect(thrownError.digest).toBe('test-error-digest');
+    }
+  });
 });

packages/react-server/src/ReactFlightServer.js

@@ -285,14 +285,18 @@ function serializeThenable(request: Request, thenable: Thenable<any>): number {
       pingTask(request, newTask);
     },
     reason => {
-      // TODO: Is it safe to directly emit these without being inside a retry?
+      newTask.status = ERRORED;
+      // TODO: We should ideally do this inside performWork so it's scheduled
       const digest = logRecoverableError(request, reason);
       if (__DEV__) {
         const {message, stack} = getErrorMessageAndStackDev(reason);
         emitErrorChunkDev(request, newTask.id, digest, message, stack);
       } else {
         emitErrorChunkProd(request, newTask.id, digest);
       }
+      if (request.destination !== null) {
+        flushCompletedChunks(request, request.destination);
+      }
     },
   );