Flush direct streams in Bun (#28837)

<!--
  Thanks for submitting a pull request!
We appreciate you spending the time to work on these changes. Please
provide enough information so that others can review your pull request.
The three fields below are mandatory.

Before submitting a pull request, please make sure the following is
done:

1. Fork [the repository](https://github.com/facebook/react) and create
your branch from `main`.
  2. Run `yarn` in the repository root.
3. If you've fixed a bug or added code that should be tested, add tests!
4. Ensure the test suite passes (`yarn test`). Tip: `yarn test --watch
TestName` is helpful in development.
5. Run `yarn test --prod` to test in the production environment. It
supports the same options as `yarn test`.
6. If you need a debugger, run `yarn test --debug --watch TestName`,
open `chrome://inspect`, and press "Inspect".
7. Format your code with
[prettier](https://github.com/prettier/prettier) (`yarn prettier`).
8. Make sure your code lints (`yarn lint`). Tip: `yarn linc` to only
check changed files.
  9. Run the [Flow](https://flowtype.org/) type checks (`yarn flow`).
  10. If you haven't already, complete the CLA.

Learn more about contributing:
https://reactjs.org/docs/how-to-contribute.html
-->

## Summary

<!--
Explain the **motivation** for making this change. What existing problem
does the pull request solve?
-->

The ReadableStreamController for [direct
streams](https://bun.sh/docs/api/streams#direct-readablestream) in Bun
supports a flush() method to flush all buffered items to its underlying
sink.

Without manually calling flush(), all buffered items are only flushed to
the underlying sink when the stream is closed. This behavior causes the
shell rendered against Suspense boundaries never to be flushed to the
underlying sink.

## How did you test this change?

<!--
Demonstrate the code is solid. Example: The exact commands you ran and
their output, screenshots / videos if the pull request changes the user
interface.
How exactly did you verify that your PR solves the issue you wanted to
solve?
  If you leave this empty, your PR will very likely be closed.
-->

A lot of changes to the test runner will need to be made in order to
support the Bun runtime. A separate test was manually run in order to
ensure that the changes made are correct.

The test works by sanity-checking that the shell rendered against
Suspense boundaries are emitted first in the stream.

This test was written and run on Bun v1.1.3.

```ts
import { Suspense } from "react";
import { renderToReadableStream } from "react-dom/server";

if (!import.meta.resolveSync("react-dom/server").endsWith("server.bun.js")) {
  throw new Error("react-dom/server is not the correct version:\n  " + import.meta.resolveSync("react-dom/server"));
}

const A = async () => {
  await new Promise(resolve => setImmediate(resolve));
  return <div>hi</div>;
};

const B = async () => {
  return (
    <Suspense fallback={<div>loading</div>}>
      <A />
    </Suspense>
  );
};

const stream = await renderToReadableStream(<B />);

let text = "";
let count = 0;
for await (const chunk of stream) {
  text += new TextDecoder().decode(chunk);
  count++;
}

if (
  text !==
  `<!--$?--><template id="B:0"></template><div>loading</div><!--/$--><div hidden id="S:0"><div>hi</div></div><script>$RC=function(b,c,e){c=document.getElementById(c);c.parentNode.removeChild(c);var a=document.getElementById(b);if(a){b=a.previousSibling;if(e)b.data="$!",a.setAttribute("data-dgst",e);else{e=b.parentNode;a=b.nextSibling;var f=0;do{if(a&&8===a.nodeType){var d=a.data;if("/$"===d)if(0===f)break;else f--;else"$"!==d&&"$?"!==d&&"$!"!==d||f++}d=a.nextSibling;e.removeChild(a);a=d}while(a);for(;c.firstChild;)e.insertBefore(c.firstChild,a);b.data="$"}b._reactRetry&&b._reactRetry()}};$RC("B:0","S:0")</script>`
) {
  throw new Error("unexpected output");
}
if (count !== 2) {
  throw new Error("expected 2 chunks from react ssr stream");
}
```

packages/react-server/src/ReactServerStreamConfigBun.js

@@ -13,6 +13,7 @@ type BunReadableStreamController = ReadableStreamController & {
   end(): mixed,
   write(data: Chunk | BinaryChunk): void,
   error(error: Error): void,
+  flush?: () => void,
 };
 export type Destination = BunReadableStreamController;
 
@@ -25,8 +26,12 @@ export function scheduleWork(callback: () => void) {
 }
 
 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
+  // Bun direct streams provide a flush function.
+  // If we don't have any more data to send right now.
+  // Flush whatever is in the buffer to the wire.
+  if (typeof destination.flush === 'function') {
+    destination.flush();
+  }
 }
 
 export function beginWriting(destination: Destination) {}