[Flight] Schedule work in a microtask (#29491)

Stacked on #29551

Flight pings much more often than Fizz because async function components
will always take at least a microtask to resolve . Rather than
scheduling this work as a new macrotask Flight now schedules pings in a
microtask. This allows more microtasks to ping before actually doing a
work flush but doesn't force the vm to spin up a new task which is quite
common give n the nature of Server Components

packages/react-dom-bindings/src/server/ReactDOMLegacyServerStreamConfig.js

@@ -20,6 +20,14 @@ export function scheduleWork(callback: () => void) {
   callback();
 }
 
+export function scheduleMicrotask(callback: () => void) {
+  // While this defies the method name the legacy builds have special
+  // overrides that make work scheduling sync. At the moment scheduleMicrotask
+  // isn't used by any legacy APIs so this is somewhat academic but if they
+  // did in the future we'd probably want to have this be in sync with scheduleWork
+  callback();
+}
+
 export function flushBuffered(destination: Destination) {}
 
 export function beginWriting(destination: Destination) {}

packages/react-noop-renderer/src/ReactNoopFlightServer.js

@@ -25,6 +25,9 @@ type Destination = Array<Uint8Array>;
 const textEncoder = new TextEncoder();
 
 const ReactNoopFlightServer = ReactFlightServer({
+  scheduleMicrotask(callback: () => void) {
+    callback();
+  },
   scheduleWork(callback: () => void) {
     callback();
   },

packages/react-noop-renderer/src/ReactNoopServer.js

@@ -74,6 +74,9 @@ function write(destination: Destination, buffer: Uint8Array): void {
 }
 
 const ReactNoopServer = ReactFizzServer({
+  scheduleMicrotask(callback: () => void) {
+    callback();
+  },
   scheduleWork(callback: () => void) {
     callback();
   },

packages/react-server/src/ReactFlightServer.js

@@ -26,6 +26,7 @@ import {enableFlightReadableStream} from 'shared/ReactFeatureFlags';
 
 import {
   scheduleWork,
+  scheduleMicrotask,
   flushBuffered,
   beginWriting,
   writeChunkAndReturn,
@@ -1571,7 +1572,7 @@ function pingTask(request: Request, task: Task): void {
   pingedTasks.push(task);
   if (pingedTasks.length === 1) {
     request.flushScheduled = request.destination !== null;
-    scheduleWork(() => performWork(request));
+    scheduleMicrotask(() => performWork(request));
   }
 }
 

packages/react-server/src/ReactServerStreamConfigBrowser.js

@@ -27,6 +27,21 @@ export function scheduleWork(callback: () => void) {
   channel.port2.postMessage(null);
 }
 
+function handleErrorInNextTick(error: any) {
+  setTimeout(() => {
+    throw error;
+  });
+}
+
+const LocalPromise = Promise;
+
+export const scheduleMicrotask: (callback: () => void) => void =
+  typeof queueMicrotask === 'function'
+    ? queueMicrotask
+    : callback => {
+        LocalPromise.resolve(null).then(callback).catch(handleErrorInNextTick);
+      };
+
 export function flushBuffered(destination: Destination) {
   // WHATWG Streams do not yet have a way to flush the underlying
   // transform streams. https://github.com/whatwg/streams/issues/960

packages/react-server/src/ReactServerStreamConfigBun.js

@@ -25,6 +25,8 @@ export function scheduleWork(callback: () => void) {
   setTimeout(callback, 0);
 }
 
+export const scheduleMicrotask = queueMicrotask;
+
 export function flushBuffered(destination: Destination) {
   // Bun direct streams provide a flush function.
   // If we don't have any more data to send right now.

packages/react-server/src/ReactServerStreamConfigEdge.js

@@ -13,6 +13,21 @@ export type PrecomputedChunk = Uint8Array;
 export opaque type Chunk = Uint8Array;
 export type BinaryChunk = Uint8Array;
 
+function handleErrorInNextTick(error: any) {
+  setTimeout(() => {
+    throw error;
+  });
+}
+
+const LocalPromise = Promise;
+
+export const scheduleMicrotask: (callback: () => void) => void =
+  typeof queueMicrotask === 'function'
+    ? queueMicrotask
+    : callback => {
+        LocalPromise.resolve(null).then(callback).catch(handleErrorInNextTick);
+      };
+
 export function scheduleWork(callback: () => void) {
   setTimeout(callback, 0);
 }

packages/react-server/src/ReactServerStreamConfigNode.js

@@ -26,6 +26,8 @@ export function scheduleWork(callback: () => void) {
   setImmediate(callback);
 }
 
+export const scheduleMicrotask = queueMicrotask;
+
 export function flushBuffered(destination: Destination) {
   // If we don't have any more data to send right now.
   // Flush whatever is in the buffer to the wire.

packages/react-server/src/forks/ReactServerStreamConfig.custom.js

@@ -31,6 +31,7 @@ export opaque type Chunk = mixed; // eslint-disable-line no-undef
 export opaque type BinaryChunk = mixed; // eslint-disable-line no-undef
 
 export const scheduleWork = $$$config.scheduleWork;
+export const scheduleMicrotask = $$$config.scheduleMicrotask;
 export const beginWriting = $$$config.beginWriting;
 export const writeChunk = $$$config.writeChunk;
 export const writeChunkAndReturn = $$$config.writeChunkAndReturn;

packages/react-server/src/forks/ReactServerStreamConfig.dom-fb-experimental.js

@@ -60,6 +60,8 @@ export function scheduleWork(callback: () => void) {
   LocalPromise.resolve().then(callback).catch(handleErrorInNextTick);
 }
 
+export const scheduleMicrotask: (callback: () => void) => void = scheduleWork;
+
 export function beginWriting(destination: Destination) {
   destination.beginWriting();
 }

packages/react-server/src/forks/ReactServerStreamConfig.dom-fb.js

@@ -9,6 +9,10 @@
 
 export * from '../ReactServerStreamConfigFB';
 
+export function scheduleMicrotask(callback: () => void) {
+  // We don't schedule work in this model, and instead expect performWork to always be called repeatedly.
+}
+
 export function scheduleWork(callback: () => void) {
   // We don't schedule work in this model, and instead expect performWork to always be called repeatedly.
 }