[Flight] Close Debug Channel when All Lazy References Have Been GC:ed (#33718)

When we have a debug channel open that can ask for more objects. That
doesn't close until all lazy objects have been explicitly asked for. If
you GC an object before the lazy references inside of it before asking
for or releasing the objects, then it'll never close.

This ensures that if there are no more PendingChunk and no more
ResolvedModelChunk then we can close the connection.

There's two sources of retaining the Response object. On one side we
have a handle to it from the stream coming from the server. On the other
side we have a handle to it from ResolvedModelChunk to ask for more data
when we lazily parse a model.

This PR makes a weak handle from the stream to the Response. However, it
keeps a strong reference alive whenever we're waiting on a pending chunk
because then the stream might be the root if the only listeners are the
callbacks passed to the promise and no references to the promise itself.

The pending chunks count can end up being zero even if we might get more
data because the references might be inside lazy chunks. In this case
the lazy chunks keeps the Response alive. When the lazy chunk gets
parsed it can find more chunks that then end up pending to keep the
response strongly alive until they resolve.

Author: sebmarkbage

fixtures/flight/src/index.js

@@ -16,18 +16,49 @@ function findSourceMapURL(fileName) {
 
 let updateRoot;
 async function callServer(id, args) {
-  const response = fetch('/', {
-    method: 'POST',
-    headers: {
-      Accept: 'text/x-component',
-      'rsc-action': id,
-    },
-    body: await encodeReply(args),
-  });
-  const {returnValue, root} = await createFromFetch(response, {
-    callServer,
-    findSourceMapURL,
-  });
+  let response;
+  if (
+    process.env.NODE_ENV === 'development' &&
+    typeof WebSocketStream === 'function'
+  ) {
+    const requestId = crypto.randomUUID();
+    const wss = new WebSocketStream(
+      'ws://localhost:3001/debug-channel?' + requestId
+    );
+    const debugChannel = await wss.opened;
+    response = createFromFetch(
+      fetch('/', {
+        method: 'POST',
+        headers: {
+          Accept: 'text/x-component',
+          'rsc-action': id,
+          'rsc-request-id': requestId,
+        },
+        body: await encodeReply(args),
+      }),
+      {
+        callServer,
+        debugChannel,
+        findSourceMapURL,
+      }
+    );
+  } else {
+    response = createFromFetch(
+      fetch('/', {
+        method: 'POST',
+        headers: {
+          Accept: 'text/x-component',
+          'rsc-action': id,
+        },
+        body: await encodeReply(args),
+      }),
+      {
+        callServer,
+        findSourceMapURL,
+      }
+    );
+  }
+  const {returnValue, root} = await response;
   // Refresh the tree with the new RSC payload.
   startTransition(() => {
     updateRoot(root);