@@ -54,13 +54,18 @@ const ReactNoopFlightServer = ReactFlightServer({
   },
 });
 
-function render(model: ReactModel): Destination {
+type Options = {
+  onError?: (error: mixed) => void,
+};
+
+function render(model: ReactModel, options?: Options): Destination {
   const destination: Destination = [];
   const bundlerConfig = undefined;
   const request = ReactNoopFlightServer.createRequest(
     model,
     destination,
     bundlerConfig,
+    options ? options.onError : undefined,
   );
   ReactNoopFlightServer.startWork(request);
   return destination;
 
@@ -15,12 +15,22 @@ import type {
 
 import {createRequest, startWork} from 'react-server/src/ReactFlightServer';
 
+type Options = {
+  onError?: (error: mixed) => void,
+};
+
 function render(
   model: ReactModel,
   destination: Destination,
   config: BundlerConfig,
+  options?: Options,
 ): void {
-  const request = createRequest(model, destination, config);
+  const request = createRequest(
+    model,
+    destination,
+    config,
+    options ? options.onError : undefined,
+  );
   startWork(request);
 }
 
 
@@ -26,6 +26,7 @@ import {resolveModelToJSON} from 'react-server/src/ReactFlightServer';
 import {
   emitRow,
   resolveModuleMetaData as resolveModuleMetaDataImpl,
+  close,
 } from 'ReactFlightDOMRelayServerIntegration';
 
 export type {
@@ -146,4 +147,8 @@ export function writeChunk(destination: Destination, chunk: Chunk): boolean {
 
 export function completeWriting(destination: Destination) {}
 
-export {close} from 'ReactFlightDOMRelayServerIntegration';
+export {close};
+
+export function closeWithError(destination: Destination, error: mixed): void {
+  close(destination);
+}
@@ -16,14 +16,24 @@ import {
   startFlowing,
 } from 'react-server/src/ReactFlightServer';
 
+type Options = {
+  onError?: (error: mixed) => void,
+};
+
 function renderToReadableStream(
   model: ReactModel,
   webpackMap: BundlerConfig,
+  options?: Options,
 ): ReadableStream {
   let request;
   return new ReadableStream({
     start(controller) {
-      request = createRequest(model, controller, webpackMap);
+      request = createRequest(
+        model,
+        controller,
+        webpackMap,
+        options ? options.onError : undefined,
+      );
       startWork(request);
     },
     pull(controller) {
 
@@ -21,12 +21,22 @@ function createDrainHandler(destination, request) {
   return () => startFlowing(request);
 }
 
+type Options = {
+  onError?: (error: mixed) => void,
+};
+
 function pipeToNodeWritable(
   model: ReactModel,
   destination: Writable,
   webpackMap: BundlerConfig,
+  options?: Options,
 ): void {
-  const request = createRequest(model, destination, webpackMap);
+  const request = createRequest(
+    model,
+    destination,
+    webpackMap,
+    options ? options.onError : undefined,
+  );
   destination.on('drain', createDrainHandler(destination, request));
   startWork(request);
 }
 
@@ -256,6 +256,7 @@ describe('ReactFlightDOM', () => {
 
   // @gate experimental
   it('should progressively reveal server components', async () => {
+    let reportedErrors = [];
     const {Suspense} = React;
 
     // Client Components
@@ -374,7 +375,11 @@ describe('ReactFlightDOM', () => {
     }
 
     const {writable, readable} = getTestStream();
-    ReactServerDOMWriter.pipeToNodeWritable(model, writable, webpackMap);
+    ReactServerDOMWriter.pipeToNodeWritable(model, writable, webpackMap, {
+      onError(x) {
+        reportedErrors.push(x);
+      },
+    });
     const response = ReactServerDOMReader.createFromReadableStream(readable);
 
     const container = document.createElement('div');
@@ -407,9 +412,12 @@ describe('ReactFlightDOM', () => {
         '<p>(loading games)</p>',
     );
 
+    expect(reportedErrors).toEqual([]);
+
+    const theError = new Error('Game over');
     // Let's *fail* loading games.
     await act(async () => {
-      rejectGames(new Error('Game over'));
+      rejectGames(theError);
     });
     expect(container.innerHTML).toBe(
       '<div>:name::avatar:</div>' +
@@ -418,6 +426,9 @@ describe('ReactFlightDOM', () => {
         '<p>Game over</p>', // TODO: should not have message in prod.
     );
 
+    expect(reportedErrors).toEqual([theError]);
+    reportedErrors = [];
+
     // We can now show the sidebar.
     await act(async () => {
       resolvePhotos();
@@ -439,6 +450,8 @@ describe('ReactFlightDOM', () => {
         '<div>:posts:</div>' +
         '<p>Game over</p>', // TODO: should not have message in prod.
     );
+
+    expect(reportedErrors).toEqual([]);
   });
 
   // @gate experimental
 
@@ -25,6 +25,7 @@ import {resolveModelToJSON} from 'react-server/src/ReactFlightServer';
 
 import {
   emitRow,
+  close,
   resolveModuleMetaData as resolveModuleMetaDataImpl,
 } from 'ReactFlightNativeRelayServerIntegration';
 
@@ -146,4 +147,8 @@ export function writeChunk(destination: Destination, chunk: Chunk): boolean {
 
 export function completeWriting(destination: Destination) {}
 
-export {close} from 'ReactFlightNativeRelayServerIntegration';
+export {close};
+
+export function closeWithError(destination: Destination, error: mixed): void {
+  close(destination);
+}
@@ -24,6 +24,7 @@ import {
   completeWriting,
   flushBuffered,
   close,
+  closeWithError,
   processModelChunk,
   processModuleChunk,
   processSymbolChunk,
@@ -83,16 +84,20 @@ export type Request = {
   completedErrorChunks: Array<Chunk>,
   writtenSymbols: Map<Symbol, number>,
   writtenModules: Map<ModuleKey, number>,
+  onError: (error: mixed) => void,
   flowing: boolean,
   toJSON: (key: string, value: ReactModel) => ReactJSONValue,
 };
 
 const ReactCurrentDispatcher = ReactSharedInternals.ReactCurrentDispatcher;
 
+function defaultErrorHandler() {}
+
 export function createRequest(
   model: ReactModel,
   destination: Destination,
   bundlerConfig: BundlerConfig,
+  onError: (error: mixed) => void = defaultErrorHandler,
 ): Request {
   const pingedSegments = [];
   const request = {
@@ -107,6 +112,7 @@ export function createRequest(
     completedErrorChunks: [],
     writtenSymbols: new Map(),
     writtenModules: new Map(),
+    onError,
     flowing: false,
     toJSON: function(key: string, value: ReactModel): ReactJSONValue {
       return resolveModelToJSON(request, this, key, value);
@@ -413,6 +419,7 @@ export function resolveModelToJSON(
         x.then(ping, ping);
         return serializeByRefID(newSegment.id);
       } else {
+        reportError(request, x);
         // Something errored. We'll still send everything we have up until this point.
         // We'll replace this element with a lazy reference that throws on the client
         // once it gets rendered.
@@ -589,6 +596,15 @@ export function resolveModelToJSON(
   );
 }
 
+function reportError(request: Request, error: mixed): void {
+  request.onError(error);
+}
+
+function fatalError(request: Request, error: mixed): void {
+  // This is called outside error handling code such as if an error happens in React internals.
+  closeWithError(request.destination, error);
+}
+
 function emitErrorChunk(request: Request, id: number, error: mixed): void {
   // TODO: We should not leak error messages to the client in prod.
   // Give this an error code instead and log on the server.
@@ -654,6 +670,7 @@ function retrySegment(request: Request, segment: Segment): void {
       x.then(ping, ping);
       return;
     } else {
+      reportError(request, x);
       // This errored, we need to serialize this error to the
       emitErrorChunk(request, segment.id, x);
     }
@@ -666,18 +683,23 @@ function performWork(request: Request): void {
   ReactCurrentDispatcher.current = Dispatcher;
   currentCache = request.cache;
 
-  const pingedSegments = request.pingedSegments;
-  request.pingedSegments = [];
-  for (let i = 0; i < pingedSegments.length; i++) {
-    const segment = pingedSegments[i];
-    retrySegment(request, segment);
-  }
-  if (request.flowing) {
-    flushCompletedChunks(request);
+  try {
+    const pingedSegments = request.pingedSegments;
+    request.pingedSegments = [];
+    for (let i = 0; i < pingedSegments.length; i++) {
+      const segment = pingedSegments[i];
+      retrySegment(request, segment);
+    }
+    if (request.flowing) {
+      flushCompletedChunks(request);
+    }
+  } catch (error) {
+    reportError(request, error);
+    fatalError(request, error);
+  } finally {
+    ReactCurrentDispatcher.current = prevDispatcher;
+    currentCache = prevCache;
   }
-
-  ReactCurrentDispatcher.current = prevDispatcher;
-  currentCache = prevCache;
 }
 
 let reentrant = false;
@@ -749,7 +771,12 @@ export function startWork(request: Request): void {
 
 export function startFlowing(request: Request): void {
   request.flowing = true;
-  flushCompletedChunks(request);
+  try {
+    flushCompletedChunks(request);
+  } catch (error) {
+    reportError(request, error);
+    fatalError(request, error);
+  }
 }
 
 function unsupportedHook(): void {
 
@@ -126,4 +126,5 @@ export {
   writeChunk,
   completeWriting,
   close,
+  closeWithError,
 } from './ReactServerStreamConfig';