@samitouri / QOS-React-2 / commits / 56389e81ff

Abort Flight (#24754)

Add aborting to the Flight Server. This encodes the reason as an "error" row that gets thrown client side. These are still exposed in prod which is a follow up we'll still have to do to encode them as digests instead. The error is encoded once and then referenced by each row that needs to be updated.

Sebastian Markbåge committed Jun 19, 2022 at 11:05 UTC 56389e81ffe17c97f3137f18cc6446a4b2c956de

8 files changed +250 -16

packages/react-server-dom-relay/src/ReactFlightDOMRelayServerHostConfig.js

@@ -117,6 +117,14 @@ export function processModelChunk(

117 return ['J', id, json];

118 }

119

120 +export function processReferenceChunk(

121 + request: Request,

122 + id: number,

123 + reference: string,

124 +): Chunk {

125 + return ['J', id, reference];

126 +}

127 +

128 export function processModuleChunk(

129 request: Request,

130 id: number,

packages/react-server-dom-webpack/src/ReactFlightDOMServerBrowser.js

+16 -2

@@ -15,12 +15,14 @@ import {

15 createRequest,

16 startWork,

17 startFlowing,

18 + abort,

19 } from 'react-server/src/ReactFlightServer';

21 type Options = {

21 - onError?: (error: mixed) => void,

22 - context?: Array<[string, ServerContextJSONValue]>,

22 identifierPrefix?: string,

23 + signal?: AbortSignal,

24 + context?: Array<[string, ServerContextJSONValue]>,

25 + onError?: (error: mixed) => void,

26 };

28 function renderToReadableStream(

@@ -35,6 +37,18 @@ function renderToReadableStream(

37 options ? options.context : undefined,

38 options ? options.identifierPrefix : undefined,

39 );

40 + if (options && options.signal) {

41 + const signal = options.signal;

42 + if (signal.aborted) {

43 + abort(request, (signal: any).reason);

44 + } else {

45 + const listener = () => {

46 + abort(request, (signal: any).reason);

47 + signal.removeEventListener('abort', listener);

48 + };

49 + signal.addEventListener('abort', listener);

50 + }

51 + }

52 const stream = new ReadableStream(

53 {

54 type: 'bytes',

packages/react-server-dom-webpack/src/ReactFlightDOMServerNode.js

@@ -16,6 +16,7 @@ import {

16 createRequest,

17 startWork,

18 startFlowing,

19 + abort,

20 } from 'react-server/src/ReactFlightServer';

22 function createDrainHandler(destination, request) {

@@ -29,6 +30,7 @@ type Options = {

30 };

32 type PipeableStream = {|

33 + abort(reason: mixed): void,

34 pipe<T: Writable>(destination: T): T,

35 |};

@@ -58,6 +60,9 @@ function renderToPipeableStream(

60 destination.on('drain', createDrainHandler(destination, request));

61 return destination;

62 },

63 + abort(reason: mixed) {

64 + abort(request, reason);

65 + },

66 };

67 }

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

+67 -9

@@ -30,6 +30,7 @@ let React;

30 let ReactDOMClient;

31 let ReactServerDOMWriter;

32 let ReactServerDOMReader;

33 +let Suspense;

35 describe('ReactFlightDOM', () => {

36 beforeEach(() => {

@@ -42,6 +43,7 @@ describe('ReactFlightDOM', () => {

43 ReactDOMClient = require('react-dom/client');

44 ReactServerDOMWriter = require('react-server-dom-webpack/writer.node.server');

45 ReactServerDOMReader = require('react-server-dom-webpack');

46 + Suspense = React.Suspense;

47 });

49 function getTestStream() {

@@ -92,6 +94,11 @@ describe('ReactFlightDOM', () => {

94 }

95 }

97 + const theInfinitePromise = new Promise(() => {});

98 + function InfiniteSuspend() {

99 + throw theInfinitePromise;

100 + }

101 +

102 it('should resolve HTML using Node streams', async () => {

103 function Text({children}) {

104 return {children};

@@ -133,8 +140,6 @@ describe('ReactFlightDOM', () => {

140 });

141

142 it('should resolve the root', async () => {

136 - const {Suspense} = React;

137 -

143 // Model

144 function Text({children}) {

145 return {children};

@@ -184,8 +189,6 @@ describe('ReactFlightDOM', () => {

189 });

190

191 it('should not get confused by $', async () => {

187 - const {Suspense} = React;

188 -

192 // Model

193 function RootModel() {

194 return {text: '$1'};

@@ -220,8 +223,6 @@ describe('ReactFlightDOM', () => {

223 });

224

225 it('should not get confused by @', async () => {

223 - const {Suspense} = React;

224 -

226 // Model

227 function RootModel() {

228 return {text: '@div'};

@@ -257,7 +258,6 @@ describe('ReactFlightDOM', () => {

258

259 it('should progressively reveal server components', async () => {

260 let reportedErrors = [];

260 - const {Suspense} = React;

261

262 // Client Components

263

@@ -460,8 +460,6 @@ describe('ReactFlightDOM', () => {

460 });

461

462 it('should preserve state of client components on refetch', async () => {

463 - const {Suspense} = React;

464 -

463 // Client

464

465 function Page({response}) {

@@ -545,4 +543,64 @@ describe('ReactFlightDOM', () => {

543 expect(inputB.tagName).toBe('INPUT');

544 expect(inputB.value).toBe('goodbye');

545 });

546 +

547 + it('should be able to complete after aborting and throw the reason client-side', async () => {

548 + const reportedErrors = [];

549 +

550 + class ErrorBoundary extends React.Component {

551 + state = {hasError: false, error: null};

552 + static getDerivedStateFromError(error) {

553 + return {

554 + hasError: true,

555 + error,

556 + };

557 + }

558 + render() {

559 + if (this.state.hasError) {

560 + return this.props.fallback(this.state.error);

561 + }

562 + return this.props.children;

563 + }

564 + }

565 +

566 + const {writable, readable} = getTestStream();

567 + const {pipe, abort} = ReactServerDOMWriter.renderToPipeableStream(

568 + <div>

569 + <InfiniteSuspend />

570 + </div>,

571 + webpackMap,

572 + {

573 + onError(x) {

574 + reportedErrors.push(x);

575 + },

576 + },

577 + );

578 + pipe(writable);

579 + const response = ReactServerDOMReader.createFromReadableStream(readable);

580 +

581 + const container = document.createElement('div');

582 + const root = ReactDOMClient.createRoot(container);

583 +

584 + function App({res}) {

585 + return res.readRoot();

586 + }

587 +

588 + await act(async () => {

589 + root.render(

590 + <ErrorBoundary fallback={e => {e.message}}>

591 + <Suspense fallback={(loading)}>

592 + <App res={response} />

593 + </Suspense>

594 + </ErrorBoundary>,

595 + );

596 + });

597 + expect(container.innerHTML).toBe('(loading)');

598 +

599 + await act(async () => {

600 + abort('for reasons');

601 + });

602 + expect(container.innerHTML).toBe('Error: for reasons');

603 +

604 + expect(reportedErrors).toEqual(['for reasons']);

605 + });

606 });

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

+71 -3

@@ -27,6 +27,7 @@ let ReactDOMClient;

27 let ReactDOMServer;

28 let ReactServerDOMWriter;

29 let ReactServerDOMReader;

30 +let Suspense;

32 describe('ReactFlightDOMBrowser', () => {

33 beforeEach(() => {

@@ -39,6 +40,7 @@ describe('ReactFlightDOMBrowser', () => {

40 ReactDOMServer = require('react-dom/server.browser');

41 ReactServerDOMWriter = require('react-server-dom-webpack/writer.browser.server');

42 ReactServerDOMReader = require('react-server-dom-webpack');

43 + Suspense = React.Suspense;

44 });

46 function moduleReference(moduleExport) {

@@ -108,6 +110,11 @@ describe('ReactFlightDOMBrowser', () => {

110 return [DelayedText, _resolve, _reject];

111 }

112

113 + const theInfinitePromise = new Promise(() => {});

114 + function InfiniteSuspend() {

115 + throw theInfinitePromise;

116 + }

117 +

118 it('should resolve HTML using W3C streams', async () => {

119 function Text({children}) {

120 return {children};

@@ -180,7 +187,6 @@ describe('ReactFlightDOMBrowser', () => {

187

188 it('should progressively reveal server components', async () => {

189 let reportedErrors = [];

183 - const {Suspense} = React;

190

191 // Client Components

192

@@ -356,8 +362,6 @@ describe('ReactFlightDOMBrowser', () => {

362 });

363

364 it('should close the stream upon completion when rendering to W3C streams', async () => {

359 - const {Suspense} = React;

360 -

365 // Model

366 function Text({children}) {

367 return children;

@@ -512,4 +516,68 @@ describe('ReactFlightDOMBrowser', () => {

516 const result = await readResult(ssrStream);

517 expect(result).toEqual('Client Component');

518 });

519 +

520 + it('should be able to complete after aborting and throw the reason client-side', async () => {

521 + const reportedErrors = [];

522 +

523 + class ErrorBoundary extends React.Component {

524 + state = {hasError: false, error: null};

525 + static getDerivedStateFromError(error) {

526 + return {

527 + hasError: true,

528 + error,

529 + };

530 + }

531 + render() {

532 + if (this.state.hasError) {

533 + return this.props.fallback(this.state.error);

534 + }

535 + return this.props.children;

536 + }

537 + }

538 +

539 + const controller = new AbortController();

540 + const stream = ReactServerDOMWriter.renderToReadableStream(

541 + <div>

542 + <InfiniteSuspend />

543 + </div>,

544 + webpackMap,

545 + {

546 + signal: controller.signal,

547 + onError(x) {

548 + reportedErrors.push(x);

549 + },

550 + },

551 + );

552 + const response = ReactServerDOMReader.createFromReadableStream(stream);

553 +

554 + const container = document.createElement('div');

555 + const root = ReactDOMClient.createRoot(container);

556 +

557 + function App({res}) {

558 + return res.readRoot();

559 + }

560 +

561 + await act(async () => {

562 + root.render(

563 + <ErrorBoundary fallback={e => {e.message}}>

564 + <Suspense fallback={(loading)}>

565 + <App res={response} />

566 + </Suspense>

567 + </ErrorBoundary>,

568 + );

569 + });

570 + expect(container.innerHTML).toBe('(loading)');

571 +

572 + await act(async () => {

573 + // @TODO this is a hack to work around lack of support for abortSignal.reason in node

574 + // The abort call itself should set this property but since we are testing in node we

575 + // set it here manually

576 + controller.signal.reason = 'for reasons';

577 + controller.abort('for reasons');

578 + });

579 + expect(container.innerHTML).toBe('Error: for reasons');

580 +

581 + expect(reportedErrors).toEqual(['for reasons']);

582 + });

583 });

packages/react-server-native-relay/src/ReactFlightNativeRelayServerHostConfig.js

@@ -114,6 +114,14 @@ export function processModelChunk(

114 return ['J', id, json];

115 }

116

117 +export function processReferenceChunk(

118 + request: Request,

119 + id: number,

120 + reference: string,

121 +): Chunk {

122 + return ['J', id, reference];

123 +}

124 +

125 export function processModuleChunk(

126 request: Request,

127 id: number,

packages/react-server/src/ReactFlightServer.js

+65 -2

@@ -34,6 +34,7 @@ import {

34 processProviderChunk,

35 processSymbolChunk,

36 processErrorChunk,

37 + processReferenceChunk,

38 resolveModuleMetaData,

39 getModuleKey,

40 isModuleReference,

@@ -86,8 +87,14 @@ export type ReactModel =

88 type ReactModelObject = {+[key: string]: ReactModel};

90 +const PENDING = 0;

91 +const COMPLETED = 1;

92 +const ABORTED = 3;

93 +const ERRORED = 4;

94 +

95 type Task = {

96 id: number,

97 + status: 0 | 1 | 3 | 4,

98 model: ReactModel,

99 ping: () => void,

100 context: ContextSnapshot,

@@ -101,6 +108,7 @@ export type Request = {

108 cache: Map<Function, mixed>,

109 nextChunkId: number,

110 pendingChunks: number,

111 + abortableTasks: Set<Task>,

112 pingedTasks: Array<Task>,

113 completedModuleChunks: Array<Chunk>,

114 completedJSONChunks: Array<Chunk>,

@@ -132,6 +140,7 @@ export function createRequest(

140 context?: Array<[string, ServerContextJSONValue]>,

141 identifierPrefix?: string,

142 ): Request {

143 + const abortSet: Set<Task> = new Set();

144 const pingedTasks = [];

145 const request = {

146 status: OPEN,

@@ -141,6 +150,7 @@ export function createRequest(

150 cache: new Map(),

151 nextChunkId: 0,

152 pendingChunks: 0,

153 + abortableTasks: abortSet,

154 pingedTasks: pingedTasks,

155 completedModuleChunks: [],

156 completedJSONChunks: [],

@@ -157,7 +167,7 @@ export function createRequest(

167 };

168 request.pendingChunks++;

169 const rootContext = createRootContext(context);

160 - const rootTask = createTask(request, model, rootContext);

170 + const rootTask = createTask(request, model, rootContext, abortSet);

171 pingedTasks.push(rootTask);

172 return request;

173 }

@@ -263,14 +273,17 @@ function createTask(

273 request: Request,

274 model: ReactModel,

275 context: ContextSnapshot,

276 + abortSet: Set<Task>,

277 ): Task {

278 const id = request.nextChunkId++;

279 const task = {

280 id,

281 + status: PENDING,

282 model,

283 context,

284 ping: () => pingTask(request, task),

285 };

286 + abortSet.add(task);

287 return task;

288 }

289

@@ -520,7 +533,12 @@ export function resolveModelToJSON(

533 if (typeof x === 'object' && x !== null && typeof x.then === 'function') {

534 // Something suspended, we'll need to create a new task and resolve it later.

535 request.pendingChunks++;

523 - const newTask = createTask(request, value, getActiveContext());

536 + const newTask = createTask(

537 + request,

538 + value,

539 + getActiveContext(),

540 + request.abortableTasks,

541 + );

542 const ping = newTask.ping;

543 x.then(ping, ping);

544 return serializeByRefID(newTask.id);

@@ -791,6 +809,10 @@ function emitProviderChunk(

809 }

810

811 function retryTask(request: Request, task: Task): void {

812 + if (task.status !== PENDING) {

813 + // We completed this by other means before we had a chance to retry it.

814 + return;

815 + }

816 switchContext(task.context);

817 try {

818 let value = task.model;

@@ -814,6 +836,8 @@ function retryTask(request: Request, task: Task): void {

836 }

837 const processedChunk = processModelChunk(request, task.id, value);

838 request.completedJSONChunks.push(processedChunk);

839 + request.abortableTasks.delete(task);

840 + task.status = COMPLETED;

841 } catch (x) {

842 if (typeof x === 'object' && x !== null && typeof x.then === 'function') {

843 // Something suspended again, let's pick it back up later.

@@ -821,6 +845,8 @@ function retryTask(request: Request, task: Task): void {

845 x.then(ping, ping);

846 return;

847 } else {

848 + request.abortableTasks.delete(task);

849 + task.status = ERRORED;

850 logRecoverableError(request, x);

851 // This errored, we need to serialize this error to the

852 emitErrorChunk(request, task.id, x);

@@ -855,6 +881,15 @@ function performWork(request: Request): void {

881 }

882 }

883

884 +function abortTask(task: Task, request: Request, errorId: number): void {

885 + task.status = ABORTED;

886 + // Instead of emitting an error per task.id, we emit a model that only

887 + // has a single value referencing the error.

888 + const ref = serializeByValueID(errorId);

889 + const processedChunk = processReferenceChunk(request, task.id, ref);

890 + request.completedJSONChunks.push(processedChunk);

891 +}

892 +

893 function flushCompletedChunks(

894 request: Request,

895 destination: Destination,

@@ -942,6 +977,34 @@ export function startFlowing(request: Request, destination: Destination): void {

977 }

978 }

979

980 +// This is called to early terminate a request. It creates an error at all pending tasks.

981 +export function abort(request: Request, reason: mixed): void {

982 + try {

983 + const abortableTasks = request.abortableTasks;

984 + if (abortableTasks.size > 0) {

985 + // We have tasks to abort. We'll emit one error row and then emit a reference

986 + // to that row from every row that's still remaining.

987 + const error =

988 + reason === undefined

989 + ? new Error('The render was aborted by the server without a reason.')

990 + : reason;

991 +

992 + logRecoverableError(request, error);

993 + request.pendingChunks++;

994 + const errorId = request.nextChunkId++;

995 + emitErrorChunk(request, errorId, error);

996 + abortableTasks.forEach(task => abortTask(task, request, errorId));

997 + abortableTasks.clear();

998 + }

999 + if (request.destination !== null) {

1000 + flushCompletedChunks(request, request.destination);

1001 + }

1002 + } catch (error) {

1003 + logRecoverableError(request, error);

1004 + fatalError(request, error);

1005 + }

1006 +}

1007 +

1008 function importServerContexts(

1009 contexts?: Array<[string, ServerContextJSONValue]>,

1010 ) {

packages/react-server/src/ReactFlightServerConfigStream.js

+10

@@ -99,6 +99,16 @@ export function processModelChunk(

99 return stringToChunk(row);

100 }

101

102 +export function processReferenceChunk(

103 + request: Request,

104 + id: number,

105 + reference: string,

106 +): Chunk {

107 + const json = stringify(reference);

108 + const row = serializeRowHeader('J', id) + json + '\n';

109 + return stringToChunk(row);

110 +}

111 +

112 export function processModuleChunk(

113 request: Request,

114 id: number,