[Flight] Use cacheController instead of abortListeners for Streams (#33633)
Now that we have `cacheSignal()` we can just use that instead of the `abortListeners` concept which was really just the same thing for cancelling the streams (ReadableStream, Blob, AsyncIterable).
Sebastian Markbåge committed
Jun 25, 2025 at 09:41 UTC
b42341ddc757129db062298f9fe3ad041c580d2a
1 file changed
+59
-93
packages/react-server/src/ReactFlightServer.js
+59
-93
@@ -433,7 +433,6 @@ export type Request = {
433
nextChunkId: number,
434
pendingChunks: number,
435
hints: Hints,
436
- abortListeners: Set<(reason: mixed) => void>,
436
abortableTasks: Set<Task>,
437
pingedTasks: Array<Task>,
438
completedImportChunks: Array<Chunk>,
@@ -547,7 +546,6 @@ function RequestInstance(
546
this.nextChunkId = 0;
547
this.pendingChunks = 0;
548
this.hints = hints;
550
- this.abortListeners = new Set();
549
this.abortableTasks = abortSet;
550
this.pingedTasks = pingedTasks;
551
this.completedImportChunks = ([]: Array<Chunk>);
@@ -839,13 +837,11 @@ function serializeThenable(
837
if (request.status === ABORTING) {
838
// We can no longer accept any resolved values
839
request.abortableTasks.delete(newTask);
842
- newTask.status = ABORTED;
840
if (enableHalt && request.type === PRERENDER) {
844
- request.pendingChunks--;
841
+ haltTask(newTask, request);
842
} else {
843
const errorId: number = (request.fatalError: any);
847
- const model = stringify(serializeByValueID(errorId));
848
- emitModelChunk(request, newTask.id, model);
844
+ abortTask(newTask, request, errorId);
845
}
846
return newTask.id;
847
}
@@ -936,29 +932,26 @@ function serializeReadableStream(
932
__DEV__ ? task.debugStack : null,
933
__DEV__ ? task.debugTask : null,
934
);
939
- request.abortableTasks.delete(streamTask);
940
-
941
- request.pendingChunks++; // The task represents the Start row. This adds a Stop row.
935
936
+ // The task represents the Stop row. This adds a Start row.
937
+ request.pendingChunks++;
938
const startStreamRow =
939
streamTask.id.toString(16) + ':' + (supportsBYOB ? 'r' : 'R') + '\n';
940
request.completedRegularChunks.push(stringToChunk(startStreamRow));
941
947
- // There's a race condition between when the stream is aborted and when the promise
948
- // resolves so we track whether we already aborted it to avoid writing twice.
949
- let aborted = false;
942
function progress(entry: {done: boolean, value: ReactClientValue, ...}) {
951
- if (aborted) {
943
+ if (streamTask.status !== PENDING) {
944
return;
945
}
946
947
if (entry.done) {
948
+ streamTask.status = COMPLETED;
949
const endStreamRow = streamTask.id.toString(16) + ':C\n';
950
request.completedRegularChunks.push(stringToChunk(endStreamRow));
951
+ request.abortableTasks.delete(streamTask);
952
+ request.cacheController.signal.removeEventListener('abort', abortStream);
953
enqueueFlush(request);
959
- request.abortListeners.delete(abortStream);
954
callOnAllReadyIfReady(request);
961
- aborted = true;
955
} else {
956
try {
957
streamTask.model = entry.value;
@@ -972,26 +965,28 @@ function serializeReadableStream(
965
}
966
}
967
function error(reason: mixed) {
975
- if (aborted) {
968
+ if (streamTask.status !== PENDING) {
969
return;
970
}
978
- aborted = true;
979
- request.abortListeners.delete(abortStream);
971
+ request.cacheController.signal.removeEventListener('abort', abortStream);
972
erroredTask(request, streamTask, reason);
973
enqueueFlush(request);
974
975
// $FlowFixMe should be able to pass mixed
976
reader.cancel(reason).then(error, error);
977
}
986
- function abortStream(reason: mixed) {
987
- if (aborted) {
978
+ function abortStream() {
979
+ if (streamTask.status !== PENDING) {
980
return;
981
}
990
- aborted = true;
991
- request.abortListeners.delete(abortStream);
982
+ const signal = request.cacheController.signal;
983
+ signal.removeEventListener('abort', abortStream);
984
+ const reason = signal.reason;
985
if (enableHalt && request.type === PRERENDER) {
993
- request.pendingChunks--;
986
+ haltTask(streamTask, request);
987
+ request.abortableTasks.delete(streamTask);
988
} else {
989
+ // TODO: Make this use abortTask() instead.
990
erroredTask(request, streamTask, reason);
991
enqueueFlush(request);
992
}
@@ -999,7 +994,7 @@ function serializeReadableStream(
994
reader.cancel(reason).then(error, error);
995
}
996
1002
- request.abortListeners.add(abortStream);
997
+ request.cacheController.signal.addEventListener('abort', abortStream);
998
reader.read().then(progress, error);
999
return serializeByValueID(streamTask.id);
1000
}
@@ -1028,10 +1023,9 @@ function serializeAsyncIterable(
1023
__DEV__ ? task.debugStack : null,
1024
__DEV__ ? task.debugTask : null,
1025
);
1031
- request.abortableTasks.delete(streamTask);
1032
-
1033
- request.pendingChunks++; // The task represents the Start row. This adds a Stop row.
1026
1027
+ // The task represents the Stop row. This adds a Start row.
1028
+ request.pendingChunks++;
1029
const startStreamRow =
1030
streamTask.id.toString(16) + ':' + (isIterator ? 'x' : 'X') + '\n';
1031
request.completedRegularChunks.push(stringToChunk(startStreamRow));
@@ -1043,19 +1037,17 @@ function serializeAsyncIterable(
1037
}
1038
}
1039
1046
- // There's a race condition between when the stream is aborted and when the promise
1047
- // resolves so we track whether we already aborted it to avoid writing twice.
1048
- let aborted = false;
1040
function progress(
1041
entry:
1042
| {done: false, +value: ReactClientValue, ...}
1043
| {done: true, +value: ReactClientValue, ...},
1044
) {
1054
- if (aborted) {
1045
+ if (streamTask.status !== PENDING) {
1046
return;
1047
}
1048
1049
if (entry.done) {
1050
+ streamTask.status = COMPLETED;
1051
let endStreamRow;
1052
if (entry.value === undefined) {
1053
endStreamRow = streamTask.id.toString(16) + ':C\n';
@@ -1075,10 +1067,13 @@ function serializeAsyncIterable(
1067
}
1068
}
1069
request.completedRegularChunks.push(stringToChunk(endStreamRow));
1070
+ request.abortableTasks.delete(streamTask);
1071
+ request.cacheController.signal.removeEventListener(
1072
+ 'abort',
1073
+ abortIterable,
1074
+ );
1075
enqueueFlush(request);
1079
- request.abortListeners.delete(abortIterable);
1076
callOnAllReadyIfReady(request);
1081
- aborted = true;
1077
} else {
1078
try {
1079
streamTask.model = entry.value;
@@ -1097,11 +1092,10 @@ function serializeAsyncIterable(
1092
}
1093
}
1094
function error(reason: mixed) {
1100
- if (aborted) {
1095
+ if (streamTask.status !== PENDING) {
1096
return;
1097
}
1103
- aborted = true;
1104
- request.abortListeners.delete(abortIterable);
1098
+ request.cacheController.signal.removeEventListener('abort', abortIterable);
1099
erroredTask(request, streamTask, reason);
1100
enqueueFlush(request);
1101
if (typeof (iterator: any).throw === 'function') {
@@ -1110,16 +1104,19 @@ function serializeAsyncIterable(
1104
iterator.throw(reason).then(error, error);
1105
}
1106
}
1113
- function abortIterable(reason: mixed) {
1114
- if (aborted) {
1107
+ function abortIterable() {
1108
+ if (streamTask.status !== PENDING) {
1109
return;
1110
}
1117
- aborted = true;
1118
- request.abortListeners.delete(abortIterable);
1111
+ const signal = request.cacheController.signal;
1112
+ signal.removeEventListener('abort', abortIterable);
1113
+ const reason = signal.reason;
1114
if (enableHalt && request.type === PRERENDER) {
1120
- request.pendingChunks--;
1115
+ haltTask(streamTask, request);
1116
+ request.abortableTasks.delete(streamTask);
1117
} else {
1122
- erroredTask(request, streamTask, reason);
1118
+ // TODO: Make this use abortTask() instead.
1119
+ erroredTask(request, streamTask, signal.reason);
1120
enqueueFlush(request);
1121
}
1122
if (typeof (iterator: any).throw === 'function') {
@@ -1128,7 +1125,7 @@ function serializeAsyncIterable(
1125
iterator.throw(reason).then(error, error);
1126
}
1127
}
1131
- request.abortListeners.add(abortIterable);
1128
+ request.cacheController.signal.addEventListener('abort', abortIterable);
1129
if (__DEV__) {
1130
callIteratorInDEV(iterator, progress, error);
1131
} else {
@@ -2675,16 +2672,14 @@ function serializeBlob(request: Request, blob: Blob): string {
2672
2673
const reader = blob.stream().getReader();
2674
2678
- let aborted = false;
2675
function progress(
2676
entry: {done: false, value: Uint8Array} | {done: true, value: void},
2677
): Promise<void> | void {
2682
- if (aborted) {
2678
+ if (newTask.status !== PENDING) {
2679
return;
2680
}
2681
if (entry.done) {
2686
- request.abortListeners.delete(abortBlob);
2687
- aborted = true;
2682
+ request.cacheController.signal.removeEventListener('abort', abortBlob);
2683
pingTask(request, newTask);
2684
return;
2685
}
@@ -2694,25 +2689,26 @@ function serializeBlob(request: Request, blob: Blob): string {
2689
return reader.read().then(progress).catch(error);
2690
}
2691
function error(reason: mixed) {
2697
- if (aborted) {
2692
+ if (newTask.status !== PENDING) {
2693
return;
2694
}
2700
- aborted = true;
2701
- request.abortListeners.delete(abortBlob);
2695
+ request.cacheController.signal.removeEventListener('abort', abortBlob);
2696
erroredTask(request, newTask, reason);
2697
enqueueFlush(request);
2698
// $FlowFixMe should be able to pass mixed
2699
reader.cancel(reason).then(error, error);
2700
}
2707
- function abortBlob(reason: mixed) {
2708
- if (aborted) {
2701
+ function abortBlob() {
2702
+ if (newTask.status !== PENDING) {
2703
return;
2704
}
2711
- aborted = true;
2712
- request.abortListeners.delete(abortBlob);
2705
+ const signal = request.cacheController.signal;
2706
+ signal.removeEventListener('abort', abortBlob);
2707
+ const reason = signal.reason;
2708
if (enableHalt && request.type === PRERENDER) {
2714
- request.pendingChunks--;
2709
+ haltTask(newTask, request);
2710
} else {
2711
+ // TODO: Make this use abortTask() instead.
2712
erroredTask(request, newTask, reason);
2713
enqueueFlush(request);
2714
}
@@ -2720,7 +2716,7 @@ function serializeBlob(request: Request, blob: Blob): string {
2716
reader.cancel(reason).then(error, error);
2717
}
2718
2723
- request.abortListeners.add(abortBlob);
2719
+ request.cacheController.signal.addEventListener('abort', abortBlob);
2720
2721
// $FlowFixMe[incompatible-call]
2722
reader.read().then(progress).catch(error);
@@ -5005,16 +5001,15 @@ function retryTask(request: Request, task: Task): void {
5001
} catch (thrownValue) {
5002
if (request.status === ABORTING) {
5003
request.abortableTasks.delete(task);
5008
- task.status = ABORTED;
5004
+ task.status = PENDING;
5005
if (enableHalt && request.type === PRERENDER) {
5006
// When aborting a prerener with halt semantics we don't emit
5007
// anything into the slot for a task that aborts, it remains unresolved
5012
- request.pendingChunks--;
5008
+ haltTask(task, request);
5009
} else {
5010
// Otherwise we emit an error chunk into the task slot.
5011
const errorId: number = (request.fatalError: any);
5016
- const model = stringify(serializeByValueID(errorId));
5017
- emitModelChunk(request, task.id, model);
5012
+ abortTask(task, request, errorId);
5013
}
5014
return;
5015
}
@@ -5257,8 +5252,9 @@ function enqueueFlush(request: Request): void {
5252
}
5253
5254
function callOnAllReadyIfReady(request: Request): void {
5260
- if (request.abortableTasks.size === 0 && request.abortListeners.size === 0) {
5261
- request.onAllReady();
5255
+ if (request.abortableTasks.size === 0) {
5256
+ const onAllReady = request.onAllReady;
5257
+ onAllReady();
5258
}
5259
}
5260
@@ -5294,6 +5290,7 @@ export function abort(request: Request, reason: mixed): void {
5290
if (request.status <= OPEN) {
5291
request.status = ABORTING;
5292
request.cacheController.abort(reason);
5293
+ callOnAllReadyIfReady(request);
5294
}
5295
const abortableTasks = request.abortableTasks;
5296
if (abortableTasks.size > 0) {
@@ -5345,37 +5342,6 @@ export function abort(request: Request, reason: mixed): void {
5342
callOnAllReadyIfReady(request);
5343
}
5344
}
5348
- const abortListeners = request.abortListeners;
5349
- if (abortListeners.size > 0) {
5350
- let error;
5351
- if (
5352
- enablePostpone &&
5353
- typeof reason === 'object' &&
5354
- reason !== null &&
5355
- (reason: any).$$typeof === REACT_POSTPONE_TYPE
5356
- ) {
5357
- // We aborted with a Postpone but since we're passing this to an
5358
- // external handler, passing this object would leak it outside React.
5359
- // We create an alternative reason for it instead.
5360
- error = new Error('The render was aborted due to being postponed.');
5361
- } else {
5362
- error =
5363
- reason === undefined
5364
- ? new Error(
5365
- 'The render was aborted by the server without a reason.',
5366
- )
5367
- : typeof reason === 'object' &&
5368
- reason !== null &&
5369
- typeof reason.then === 'function'
5370
- ? new Error(
5371
- 'The render was aborted by the server with a promise.',
5372
- )
5373
- : reason;
5374
- }
5375
- abortListeners.forEach(callback => callback(error));
5376
- abortListeners.clear();
5377
- callOnAllReadyIfReady(request);
5378
- }
5345
if (request.destination !== null) {
5346
flushCompletedChunks(request, request.destination);
5347
}