[Flight Reply] Align Rspack decoders with upstream changes

.eslintrc.js

@@ -589,6 +589,8 @@ module.exports = {
  CopyInspectedElementPath: 'readonly',
  DOMHighResTimeStamp: 'readonly',
  EventListener: 'readonly',
+ // Flow type
+ FormDataEntryValue: 'readonly',
  Iterable: 'readonly',
  AsyncIterable: 'readonly',
  $AsyncIterable: 'readonly',

flow-typed/environments/bom.js

@@ -682,11 +682,11 @@ declare class FormData {
  get(name: string): ?FormDataEntryValue;
  getAll(name: string): Array<FormDataEntryValue>;
 
- set(name: string, value: string): void;
+ set(name: string, value: FormDataEntryValue): void;
  set(name: string, value: Blob, filename?: string): void;
  set(name: string, value: File, filename?: string): void;
 
- append(name: string, value: string): void;
+ append(name: string, value: FormDataEntryValue): void;
  append(name: string, value: Blob, filename?: string): void;
  append(name: string, value: File, filename?: string): void;
 

package.json

@@ -53,6 +53,7 @@
  "art": "0.10.1",
  "babel-plugin-syntax-hermes-parser": "^0.32.0",
  "babel-plugin-syntax-trailing-function-commas": "^6.5.0",
+ "busboy": "^1.6.0",
  "chalk": "^3.0.0",
  "cli-table": "^0.3.1",
  "coffee-script": "^1.12.7",

packages/react-client/src/ReactFlightReplyClient.js

@@ -590,7 +590,9 @@ export function processReply(
  // Copy all the form fields with a prefix for this reference.
  // These must come first in the form order because we assume that all the
  // fields are available before this is referenced.
- const prefix = formFieldPrefix + refId + '_';
+ // We include a special marker so that the Server can detect FormData entries
+ // that are values in referenced FormData objects.
+ const prefix = formFieldPrefix + '_' + refId + '_';
  // $FlowFixMe[prop-missing]: FormData has forEach.
  value.forEach((originalValue: string | File, originalKey: string) => {
  // $FlowFixMe[incompatible-call]

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

@@ -62,6 +62,7 @@ import {
 } from 'react-client/src/ReactFlightClientStreamConfigNode';
 
 import type {TemporaryReferenceSet} from 'react-server/src/ReactFlightServerTemporaryReferences';
+import type {FileHandle} from 'react-server/src/ReactFlightReplyServer';
 
 export {createTemporaryReferenceSet} from 'react-server/src/ReactFlightServerTemporaryReferences';
 
@@ -325,6 +326,17 @@ function prerenderToNodeStream(
  });
 }
 
+type PendingFile = {
+ name: string,
+ file: FileHandle,
+ complete: boolean,
+ // Lazily allocated when a text field arrives after this file's 'file'
+ // event but before its (deferred) 'end' event. Stored as flat
+ // [name1, value1, name2, value2, ...] pairs.
+ queuedFields: null | Array<string>,
+ next: null | PendingFile,
+};
+
 function decodeReplyFromBusboy<T>(
  busboyStream: Busboy,
  moduleBasePath: ServerManifest,
@@ -340,14 +352,55 @@ function decodeReplyFromBusboy<T>(
  undefined,
  options ? options.arraySizeLimit : undefined,
  );
- let pendingFiles = 0;
- const queuedFields: Array<string> = [];
+
+ // Linked list of pending files in arrival (payload) order. Text fields that
+ // arrive while a file is in flight are queued on the tail file's
+ // `queuedFields` so they can be resolved together when that file completes.
+ // Fields that arrive while the list is empty bypass it and resolve
+ // immediately. This makes the backing FormData's insertion order match the
+ // payload's entry order.
+ let head: null | PendingFile = null;
+ let tail: null | PendingFile = null;
+ let bodyFinished = false;
+ let closed = false;
+
+ function flush() {
+ while (head !== null) {
+ const current = head;
+ if (!current.complete) {
+ // This file is still streaming. Hold later files and fields until it
+ // completes so the backing FormData reflects payload order.
+ return;
+ }
+ try {
+ resolveFileComplete(response, current.name, current.file);
+ const queuedFields = current.queuedFields;
+ if (queuedFields !== null) {
+ for (let i = 0; i < queuedFields.length; i += 2) {
+ resolveField(response, queuedFields[i], queuedFields[i + 1]);
+ }
+ }
+ } catch (error) {
+ busboyStream.destroy(error);
+ return;
+ }
+ head = current.next;
+ }
+ tail = null;
+ if (bodyFinished && !closed) {
+ closed = true;
+ close(response);
+ }
+ }
+
  busboyStream.on('field', (name, value) => {
- if (pendingFiles > 0) {
- // Because the 'end' event fires two microtasks after the next 'field'
- // we would resolve files and fields out of order. To handle this properly
- // we queue any fields we receive until the previous file is done.
- queuedFields.push(name, value);
+ if (tail !== null) {
+ // A file is in flight; queue the field on the tail (most recent) pending
+ // file so it resolves after that file, preserving payload order.
+ if (tail.queuedFields === null) {
+ tail.queuedFields = [];
+ }
+ tail.queuedFields.push(name, value);
  } else {
  try {
  resolveField(response, name, value);
@@ -367,29 +420,46 @@ function decodeReplyFromBusboy<T>(
  );
  return;
  }
- pendingFiles++;
  const file = resolveFileInfo(response, name, filename, mimeType);
+ const pendingFile: PendingFile = {
+ name,
+ file,
+ complete: false,
+ queuedFields: null,
+ next: null,
+ };
+ if (tail === null) {
+ head = pendingFile;
+ } else {
+ tail.next = pendingFile;
+ }
+ tail = pendingFile;
  value.on('data', chunk => {
- resolveFileChunk(response, file, chunk);
- });
- value.on('end', () => {
  try {
- resolveFileComplete(response, name, file);
- pendingFiles--;
- if (pendingFiles === 0) {
- // Release any queued fields
- for (let i = 0; i < queuedFields.length; i += 2) {
- resolveField(response, queuedFields[i], queuedFields[i + 1]);
- }
- queuedFields.length = 0;
- }
+ resolveFileChunk(response, file, chunk);
  } catch (error) {
  busboyStream.destroy(error);
  }
  });
+ value.on('error', error => {
+ busboyStream.destroy(error);
+ });
+ value.on('end', () => {
+ pendingFile.complete = true;
+ flush();
+ });
  });
  busboyStream.on('finish', () => {
- close(response);
+ bodyFinished = true;
+ flush();
+ if (!closed) {
+ // Invariant: busboy delays 'finish' until every file's 'end' event has
+ // fired, so the flush above should always close the response.
+ reportGlobalError(
+ response,
+ new Error('Reply finished with incomplete file part.'),
+ );
+ }
  });
  busboyStream.on('error', err => {
  reportGlobalError(

packages/react-server-dom-parcel/src/server/ReactFlightDOMServerEdge.js

@@ -20,6 +20,7 @@ import {
  type ServerReferenceId,
 } from '../client/ReactFlightClientConfigBundlerParcel';
 
+import noop from 'shared/noop';
 import {ASYNC_ITERATOR} from 'shared/ReactSymbols';
 
 import {
@@ -306,8 +307,8 @@ export function decodeReplyFromAsyncIterable<T>(
  reportGlobalError(response, reason);
  if (typeof (iterator: any).throw === 'function') {
  // The iterator protocol doesn't necessarily include this but a generator do.
- // $FlowFixMe should be able to pass mixed
- iterator.throw(reason).then(error, error);
+ // $FlowFixMe[prop-missing] should be able to pass mixed
+ iterator.throw(reason).then(noop, noop);
  }
  }
 

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

@@ -24,6 +24,7 @@ import type {Duplex} from 'stream';
 
 import {Readable} from 'stream';
 
+import noop from 'shared/noop';
 import {ASYNC_ITERATOR} from 'shared/ReactSymbols';
 
 import {
@@ -75,6 +76,7 @@ import {
 import {textEncoder} from 'react-server/src/ReactServerStreamConfigNode';
 
 import type {TemporaryReferenceSet} from 'react-server/src/ReactFlightServerTemporaryReferences';
+import type {FileHandle} from 'react-server/src/ReactFlightReplyServer';
 
 export {createTemporaryReferenceSet} from 'react-server/src/ReactFlightServerTemporaryReferences';
 
@@ -554,6 +556,17 @@ export function registerServerActions(manifest: ServerManifest) {
  serverManifest = manifest;
 }
 
+type PendingFile = {
+ name: string,
+ file: FileHandle,
+ complete: boolean,
+ // Lazily allocated when a text field arrives after this file's 'file'
+ // event but before its (deferred) 'end' event. Stored as flat
+ // [name1, value1, name2, value2, ...] pairs.
+ queuedFields: null | Array<string>,
+ next: null | PendingFile,
+};
+
 export function decodeReplyFromBusboy<T>(
  busboyStream: Busboy,
  options?: {
@@ -568,14 +581,55 @@ export function decodeReplyFromBusboy<T>(
  undefined,
  options ? options.arraySizeLimit : undefined,
  );
- let pendingFiles = 0;
- const queuedFields: Array<string> = [];
+
+ // Linked list of pending files in arrival (payload) order. Text fields that
+ // arrive while a file is in flight are queued on the tail file's
+ // `queuedFields` so they can be resolved together when that file completes.
+ // Fields that arrive while the list is empty bypass it and resolve
+ // immediately. This makes the backing FormData's insertion order match the
+ // payload's entry order.
+ let head: null | PendingFile = null;
+ let tail: null | PendingFile = null;
+ let bodyFinished = false;
+ let closed = false;
+
+ function flush() {
+ while (head !== null) {
+ const current = head;
+ if (!current.complete) {
+ // This file is still streaming. Hold later files and fields until it
+ // completes so the backing FormData reflects payload order.
+ return;
+ }
+ try {
+ resolveFileComplete(response, current.name, current.file);
+ const queuedFields = current.queuedFields;
+ if (queuedFields !== null) {
+ for (let i = 0; i < queuedFields.length; i += 2) {
+ resolveField(response, queuedFields[i], queuedFields[i + 1]);
+ }
+ }
+ } catch (error) {
+ busboyStream.destroy(error);
+ return;
+ }
+ head = current.next;
+ }
+ tail = null;
+ if (bodyFinished && !closed) {
+ closed = true;
+ close(response);
+ }
+ }
+
  busboyStream.on('field', (name, value) => {
- if (pendingFiles > 0) {
- // Because the 'end' event fires two microtasks after the next 'field'
- // we would resolve files and fields out of order. To handle this properly
- // we queue any fields we receive until the previous file is done.
- queuedFields.push(name, value);
+ if (tail !== null) {
+ // A file is in flight; queue the field on the tail (most recent) pending
+ // file so it resolves after that file, preserving payload order.
+ if (tail.queuedFields === null) {
+ tail.queuedFields = [];
+ }
+ tail.queuedFields.push(name, value);
  } else {
  try {
  resolveField(response, name, value);
@@ -595,29 +649,46 @@ export function decodeReplyFromBusboy<T>(
  );
  return;
  }
- pendingFiles++;
  const file = resolveFileInfo(response, name, filename, mimeType);
+ const pendingFile: PendingFile = {
+ name,
+ file,
+ complete: false,
+ queuedFields: null,
+ next: null,
+ };
+ if (tail === null) {
+ head = pendingFile;
+ } else {
+ tail.next = pendingFile;
+ }
+ tail = pendingFile;
  value.on('data', chunk => {
- resolveFileChunk(response, file, chunk);
- });
- value.on('end', () => {
  try {
- resolveFileComplete(response, name, file);
- pendingFiles--;
- if (pendingFiles === 0) {
- // Release any queued fields
- for (let i = 0; i < queuedFields.length; i += 2) {
- resolveField(response, queuedFields[i], queuedFields[i + 1]);
- }
- queuedFields.length = 0;
- }
+ resolveFileChunk(response, file, chunk);
  } catch (error) {
  busboyStream.destroy(error);
  }
  });
+ value.on('error', error => {
+ busboyStream.destroy(error);
+ });
+ value.on('end', () => {
+ pendingFile.complete = true;
+ flush();
+ });
  });
  busboyStream.on('finish', () => {
- close(response);
+ bodyFinished = true;
+ flush();
+ if (!closed) {
+ // Invariant: busboy delays 'finish' until every file's 'end' event has
+ // fired, so the flush above should always close the response.
+ reportGlobalError(
+ response,
+ new Error('Reply finished with incomplete file part.'),
+ );
+ }
  });
  busboyStream.on('error', err => {
  reportGlobalError(
@@ -692,8 +763,8 @@ export function decodeReplyFromAsyncIterable<T>(
  reportGlobalError(response, reason);
  if (typeof (iterator: any).throw === 'function') {
  // The iterator protocol doesn't necessarily include this but a generator do.
- // $FlowFixMe should be able to pass mixed
- iterator.throw(reason).then(error, error);
+ // $FlowFixMe[prop-missing] should be able to pass mixed
+ iterator.throw(reason).then(noop, noop);
  }
  }