fix: close fetch response body on GC (#11467)

This commit fixes fetch response bodies to be automatically closed if
the `Response.body` readable stream goes out of scope and is GC'ed.

cli/tests/fetch_response_finalization.js

@@ -0,0 +1,16 @@
+async function doAFetch() {
+  const resp = await fetch("http://localhost:4545/README.md");
+  console.log(Deno.resources()); // print the current resources
+  const _resp = resp;
+  // at this point resp can be GC'ed
+}
+
+await doAFetch(); // create a resource
+
+globalThis.gc(); // force GC
+
+// It is very important that there is a yield here, otherwise the finalizer for
+// the response body is not called and the resource is not closed.
+await new Promise((resolve) => setTimeout(resolve, 0));
+
+console.log(Deno.resources()); // print the current resources

cli/tests/fetch_response_finalization.js.out

@@ -0,0 +1,2 @@
+{ "0": "stdin", "1": "stdout", "2": "stderr", "5": "fetchResponseBody" }
+{ "0": "stdin", "1": "stdout", "2": "stderr" }

cli/tests/integration/run_tests.rs

@@ -372,6 +372,13 @@ itest!(blob_gc_finalization {
   exit_code: 0,
 });
 
+itest!(fetch_response_finalization {
+  args: "run --v8-flags=--expose-gc --allow-net fetch_response_finalization.js",
+  output: "fetch_response_finalization.js.out",
+  http_server: true,
+  exit_code: 0,
+});
+
 itest!(lock_write_requires_lock {
   args: "run --lock-write some_file.ts",
   output: "lock_write_requires_lock.out",

extensions/fetch/26_fetch.js

@@ -83,6 +83,15 @@
     return core.opAsync("op_fetch_response_read", rid, body);
   }
 
+  // A finalization registry to clean up underlying fetch resources that are GC'ed.
+  const RESOURCE_REGISTRY = new FinalizationRegistry((rid) => {
+    try {
+      core.close(rid);
+    } catch {
+      // might have already been closed
+    }
+  });
+
   /**
    * @param {number} responseBodyRid
    * @param {AbortSignal} [terminator]
@@ -119,6 +128,7 @@
             // We read some data. Enqueue it onto the stream.
             controller.enqueue(TypedArrayPrototypeSubarray(chunk, 0, read));
           } else {
+            RESOURCE_REGISTRY.unregister(readable);
             // We have reached the end of the body, so we close the stream.
             controller.close();
             try {
@@ -128,6 +138,7 @@
             }
           }
         } catch (err) {
+          RESOURCE_REGISTRY.unregister(readable);
           if (terminator.aborted) {
             controller.error(
               new DOMException("Ongoing fetch was aborted.", "AbortError"),
@@ -150,6 +161,7 @@
         }
       },
     });
+    RESOURCE_REGISTRY.register(readable, responseBodyRid, readable);
     return readable;
   }