fix: panic in request body streaming (#11191)

cli/tests/unit/fetch_test.ts

@@ -3,6 +3,7 @@ import {
   assert,
   assertEquals,
   assertThrowsAsync,
+  deferred,
   fail,
   unimplemented,
   unitTest,
@@ -1195,3 +1196,24 @@ unitTest(
     assertEquals(response.headers.get("Host"), addr);
   },
 );
+
+unitTest(
+  { perms: { net: true } },
+  async function fetchNoServerReadableStreamBody() {
+    const done = deferred();
+    const body = new ReadableStream({
+      start(controller) {
+        controller.enqueue(new Uint8Array([1]));
+        setTimeout(() => {
+          controller.enqueue(new Uint8Array([2]));
+          done.resolve();
+        }, 1000);
+      },
+    });
+    const nonExistantHostname = "http://localhost:47582";
+    await assertThrowsAsync(async () => {
+      await fetch(nonExistantHostname, { body, method: "POST" });
+    }, TypeError);
+    await done;
+  },
+);

extensions/fetch/lib.rs

@@ -356,7 +356,9 @@ pub async fn op_fetch_request_write(
     .ok_or_else(bad_resource_id)?;
   let body = RcRef::map(&resource, |r| &r.body).borrow_mut().await;
   let cancel = RcRef::map(resource, |r| &r.cancel);
-  body.send(Ok(buf)).or_cancel(cancel).await??;
+  body.send(Ok(buf)).or_cancel(cancel).await?.map_err(|_| {
+    type_error("request body receiver not connected (request closed)")
+  })?;
 
   Ok(())
 }