[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);
     }
   }