Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Abort Flight #24754

Merged
merged 1 commit into from
Jun 19, 2022
Merged
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Original file line number Diff line number Diff line change
@@ -117,6 +117,14 @@ export function processModelChunk(
return ['J', id, json];
}

export function processReferenceChunk(
request: Request,
id: number,
reference: string,
): Chunk {
return ['J', id, reference];
}

export function processModuleChunk(
request: Request,
id: number,
Original file line number Diff line number Diff line change
@@ -15,12 +15,14 @@ import {
createRequest,
startWork,
startFlowing,
abort,
} from 'react-server/src/ReactFlightServer';

type Options = {
onError?: (error: mixed) => void,
context?: Array<[string, ServerContextJSONValue]>,
identifierPrefix?: string,
signal?: AbortSignal,
context?: Array<[string, ServerContextJSONValue]>,
onError?: (error: mixed) => void,
};

function renderToReadableStream(
@@ -35,6 +37,18 @@ function renderToReadableStream(
options ? options.context : undefined,
options ? options.identifierPrefix : undefined,
);
if (options && options.signal) {
const signal = options.signal;
if (signal.aborted) {
abort(request, (signal: any).reason);
} else {
const listener = () => {
abort(request, (signal: any).reason);
signal.removeEventListener('abort', listener);
};
signal.addEventListener('abort', listener);
}
}
const stream = new ReadableStream(
{
type: 'bytes',
Original file line number Diff line number Diff line change
@@ -16,6 +16,7 @@ import {
createRequest,
startWork,
startFlowing,
abort,
} from 'react-server/src/ReactFlightServer';

function createDrainHandler(destination, request) {
@@ -29,6 +30,7 @@ type Options = {
};

type PipeableStream = {|
abort(reason: mixed): void,
pipe<T: Writable>(destination: T): T,
|};

@@ -58,6 +60,9 @@ function renderToPipeableStream(
destination.on('drain', createDrainHandler(destination, request));
return destination;
},
abort(reason: mixed) {
abort(request, reason);
},
};
}

Original file line number Diff line number Diff line change
@@ -30,6 +30,7 @@ let React;
let ReactDOMClient;
let ReactServerDOMWriter;
let ReactServerDOMReader;
let Suspense;

describe('ReactFlightDOM', () => {
beforeEach(() => {
@@ -42,6 +43,7 @@ describe('ReactFlightDOM', () => {
ReactDOMClient = require('react-dom/client');
ReactServerDOMWriter = require('react-server-dom-webpack/writer.node.server');
ReactServerDOMReader = require('react-server-dom-webpack');
Suspense = React.Suspense;
});

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

const theInfinitePromise = new Promise(() => {});
function InfiniteSuspend() {
throw theInfinitePromise;
}

it('should resolve HTML using Node streams', async () => {
function Text({children}) {
return <span>{children}</span>;
@@ -133,8 +140,6 @@ describe('ReactFlightDOM', () => {
});

it('should resolve the root', async () => {
const {Suspense} = React;

// Model
function Text({children}) {
return <span>{children}</span>;
@@ -184,8 +189,6 @@ describe('ReactFlightDOM', () => {
});

it('should not get confused by $', async () => {
const {Suspense} = React;

// Model
function RootModel() {
return {text: '$1'};
@@ -220,8 +223,6 @@ describe('ReactFlightDOM', () => {
});

it('should not get confused by @', async () => {
const {Suspense} = React;

// Model
function RootModel() {
return {text: '@div'};
@@ -257,7 +258,6 @@ describe('ReactFlightDOM', () => {

it('should progressively reveal server components', async () => {
let reportedErrors = [];
const {Suspense} = React;

// Client Components

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

it('should preserve state of client components on refetch', async () => {
const {Suspense} = React;

// Client

function Page({response}) {
@@ -545,4 +543,64 @@ describe('ReactFlightDOM', () => {
expect(inputB.tagName).toBe('INPUT');
expect(inputB.value).toBe('goodbye');
});

it('should be able to complete after aborting and throw the reason client-side', async () => {
const reportedErrors = [];

class ErrorBoundary extends React.Component {
state = {hasError: false, error: null};
static getDerivedStateFromError(error) {
return {
hasError: true,
error,
};
}
render() {
if (this.state.hasError) {
return this.props.fallback(this.state.error);
}
return this.props.children;
}
}

const {writable, readable} = getTestStream();
const {pipe, abort} = ReactServerDOMWriter.renderToPipeableStream(
<div>
<InfiniteSuspend />
</div>,
webpackMap,
{
onError(x) {
reportedErrors.push(x);
},
},
);
pipe(writable);
const response = ReactServerDOMReader.createFromReadableStream(readable);

const container = document.createElement('div');
const root = ReactDOMClient.createRoot(container);

function App({res}) {
return res.readRoot();
}

await act(async () => {
root.render(
<ErrorBoundary fallback={e => <p>{e.message}</p>}>
<Suspense fallback={<p>(loading)</p>}>
<App res={response} />
</Suspense>
</ErrorBoundary>,
);
});
expect(container.innerHTML).toBe('<p>(loading)</p>');

await act(async () => {
abort('for reasons');
});
expect(container.innerHTML).toBe('<p>Error: for reasons</p>');

expect(reportedErrors).toEqual(['for reasons']);
});
});
Original file line number Diff line number Diff line change
@@ -27,6 +27,7 @@ let ReactDOMClient;
let ReactDOMServer;
let ReactServerDOMWriter;
let ReactServerDOMReader;
let Suspense;

describe('ReactFlightDOMBrowser', () => {
beforeEach(() => {
@@ -39,6 +40,7 @@ describe('ReactFlightDOMBrowser', () => {
ReactDOMServer = require('react-dom/server.browser');
ReactServerDOMWriter = require('react-server-dom-webpack/writer.browser.server');
ReactServerDOMReader = require('react-server-dom-webpack');
Suspense = React.Suspense;
});

function moduleReference(moduleExport) {
@@ -108,6 +110,11 @@ describe('ReactFlightDOMBrowser', () => {
return [DelayedText, _resolve, _reject];
}

const theInfinitePromise = new Promise(() => {});
function InfiniteSuspend() {
throw theInfinitePromise;
}

it('should resolve HTML using W3C streams', async () => {
function Text({children}) {
return <span>{children}</span>;
@@ -180,7 +187,6 @@ describe('ReactFlightDOMBrowser', () => {

it('should progressively reveal server components', async () => {
let reportedErrors = [];
const {Suspense} = React;

// Client Components

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

it('should close the stream upon completion when rendering to W3C streams', async () => {
const {Suspense} = React;

// Model
function Text({children}) {
return children;
@@ -512,4 +516,68 @@ describe('ReactFlightDOMBrowser', () => {
const result = await readResult(ssrStream);
expect(result).toEqual('<span>Client Component</span>');
});

it('should be able to complete after aborting and throw the reason client-side', async () => {
const reportedErrors = [];

class ErrorBoundary extends React.Component {
state = {hasError: false, error: null};
static getDerivedStateFromError(error) {
return {
hasError: true,
error,
};
}
render() {
if (this.state.hasError) {
return this.props.fallback(this.state.error);
}
return this.props.children;
}
}

const controller = new AbortController();
const stream = ReactServerDOMWriter.renderToReadableStream(
<div>
<InfiniteSuspend />
</div>,
webpackMap,
{
signal: controller.signal,
onError(x) {
reportedErrors.push(x);
},
},
);
const response = ReactServerDOMReader.createFromReadableStream(stream);

const container = document.createElement('div');
const root = ReactDOMClient.createRoot(container);

function App({res}) {
return res.readRoot();
}

await act(async () => {
root.render(
<ErrorBoundary fallback={e => <p>{e.message}</p>}>
<Suspense fallback={<p>(loading)</p>}>
<App res={response} />
</Suspense>
</ErrorBoundary>,
);
});
expect(container.innerHTML).toBe('<p>(loading)</p>');

await act(async () => {
// @TODO this is a hack to work around lack of support for abortSignal.reason in node
// The abort call itself should set this property but since we are testing in node we
// set it here manually
controller.signal.reason = 'for reasons';
controller.abort('for reasons');
});
expect(container.innerHTML).toBe('<p>Error: for reasons</p>');

expect(reportedErrors).toEqual(['for reasons']);
});
});
Original file line number Diff line number Diff line change
@@ -114,6 +114,14 @@ export function processModelChunk(
return ['J', id, json];
}

export function processReferenceChunk(
request: Request,
id: number,
reference: string,
): Chunk {
return ['J', id, reference];
}

export function processModuleChunk(
request: Request,
id: number,
67 changes: 65 additions & 2 deletions packages/react-server/src/ReactFlightServer.js
Original file line number Diff line number Diff line change
@@ -34,6 +34,7 @@ import {
processProviderChunk,
processSymbolChunk,
processErrorChunk,
processReferenceChunk,
resolveModuleMetaData,
getModuleKey,
isModuleReference,
@@ -86,8 +87,14 @@ export type ReactModel =

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

const PENDING = 0;
const COMPLETED = 1;
const ABORTED = 3;
const ERRORED = 4;

type Task = {
id: number,
status: 0 | 1 | 3 | 4,
model: ReactModel,
ping: () => void,
context: ContextSnapshot,
@@ -101,6 +108,7 @@ export type Request = {
cache: Map<Function, mixed>,
nextChunkId: number,
pendingChunks: number,
abortableTasks: Set<Task>,
pingedTasks: Array<Task>,
completedModuleChunks: Array<Chunk>,
completedJSONChunks: Array<Chunk>,
@@ -132,6 +140,7 @@ export function createRequest(
context?: Array<[string, ServerContextJSONValue]>,
identifierPrefix?: string,
): Request {
const abortSet: Set<Task> = new Set();
const pingedTasks = [];
const request = {
status: OPEN,
@@ -141,6 +150,7 @@ export function createRequest(
cache: new Map(),
nextChunkId: 0,
pendingChunks: 0,
abortableTasks: abortSet,
pingedTasks: pingedTasks,
completedModuleChunks: [],
completedJSONChunks: [],
@@ -157,7 +167,7 @@ export function createRequest(
};
request.pendingChunks++;
const rootContext = createRootContext(context);
const rootTask = createTask(request, model, rootContext);
const rootTask = createTask(request, model, rootContext, abortSet);
pingedTasks.push(rootTask);
return request;
}
@@ -263,14 +273,17 @@ function createTask(
request: Request,
model: ReactModel,
context: ContextSnapshot,
abortSet: Set<Task>,
): Task {
const id = request.nextChunkId++;
const task = {
id,
status: PENDING,
model,
context,
ping: () => pingTask(request, task),
};
abortSet.add(task);
return task;
}

@@ -520,7 +533,12 @@ export function resolveModelToJSON(
if (typeof x === 'object' && x !== null && typeof x.then === 'function') {
// Something suspended, we'll need to create a new task and resolve it later.
request.pendingChunks++;
const newTask = createTask(request, value, getActiveContext());
const newTask = createTask(
request,
value,
getActiveContext(),
request.abortableTasks,
);
const ping = newTask.ping;
x.then(ping, ping);
return serializeByRefID(newTask.id);
@@ -791,6 +809,10 @@ function emitProviderChunk(
}

function retryTask(request: Request, task: Task): void {
if (task.status !== PENDING) {
// We completed this by other means before we had a chance to retry it.
return;
}
switchContext(task.context);
try {
let value = task.model;
@@ -814,13 +836,17 @@ function retryTask(request: Request, task: Task): void {
}
const processedChunk = processModelChunk(request, task.id, value);
request.completedJSONChunks.push(processedChunk);
request.abortableTasks.delete(task);
task.status = COMPLETED;
} catch (x) {
if (typeof x === 'object' && x !== null && typeof x.then === 'function') {
// Something suspended again, let's pick it back up later.
const ping = task.ping;
x.then(ping, ping);
return;
} else {
request.abortableTasks.delete(task);
task.status = ERRORED;
logRecoverableError(request, x);
// This errored, we need to serialize this error to the
emitErrorChunk(request, task.id, x);
@@ -855,6 +881,15 @@ function performWork(request: Request): void {
}
}

function abortTask(task: Task, request: Request, errorId: number): void {
task.status = ABORTED;
// Instead of emitting an error per task.id, we emit a model that only
// has a single value referencing the error.
const ref = serializeByValueID(errorId);
const processedChunk = processReferenceChunk(request, task.id, ref);
request.completedJSONChunks.push(processedChunk);
}

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

// This is called to early terminate a request. It creates an error at all pending tasks.
export function abort(request: Request, reason: mixed): void {
try {
const abortableTasks = request.abortableTasks;
if (abortableTasks.size > 0) {
// We have tasks to abort. We'll emit one error row and then emit a reference
// to that row from every row that's still remaining.
const error =
reason === undefined
? new Error('The render was aborted by the server without a reason.')
: reason;

logRecoverableError(request, error);
request.pendingChunks++;
const errorId = request.nextChunkId++;
emitErrorChunk(request, errorId, error);
abortableTasks.forEach(task => abortTask(task, request, errorId));
abortableTasks.clear();
}
if (request.destination !== null) {
flushCompletedChunks(request, request.destination);
}
} catch (error) {
logRecoverableError(request, error);
fatalError(request, error);
}
}

function importServerContexts(
contexts?: Array<[string, ServerContextJSONValue]>,
) {
10 changes: 10 additions & 0 deletions packages/react-server/src/ReactFlightServerConfigStream.js
Original file line number Diff line number Diff line change
@@ -99,6 +99,16 @@ export function processModelChunk(
return stringToChunk(row);
}

export function processReferenceChunk(
request: Request,
id: number,
reference: string,
): Chunk {
const json = stringify(reference);
const row = serializeRowHeader('J', id) + json + '\n';
return stringToChunk(row);
}

export function processModuleChunk(
request: Request,
id: number,