feat(ext/web): use readableStreamDefaultReaderRead in resourceForReadableStream (#20622)

We can go one level down in abstraction and avoid using the public
`ReadableStream` APIs.

This patch ~5% perf boost on small ReadableStream:

```
Running 10s test @ http://localhost:8080/
  2 threads and 10 connections
  Thread Stats   Avg      Stdev     Max   +/- Stdev
    Latency   148.32us  108.95us   3.88ms   95.71%
    Req/Sec    33.24k     2.68k   37.94k    73.76%
  668188 requests in 10.10s, 77.74MB read
Requests/sec:  66162.91
Transfer/sec:      7.70MB
```

main:

```
Running 10s test @ http://localhost:8080/
  2 threads and 10 connections
  Thread Stats   Avg      Stdev     Max   +/- Stdev
    Latency   150.23us   67.61us   4.39ms   94.80%
    Req/Sec    31.81k     1.55k   35.56k    83.17%
  639078 requests in 10.10s, 74.36MB read
Requests/sec:  63273.72
Transfer/sec:      7.36MB
```

1 files changed, 29 insertions, 8 deletions

diff --git a/ext/web/stream_resource.rs b/ext/web/stream_resource.rs
index 1ee6ff963..e19954fdc 100644
--- a/ext/web/stream_resource.rs
+++ b/ext/web/stream_resource.rs
@@ -313,6 +313,10 @@ impl BoundedBufferChannel {
     self.inner().write_error(error)
   }
 
+  pub fn can_write(&self) -> bool {
+    self.inner().can_write()
+  }
+
   pub fn poll_read_ready(&self, cx: &mut Context) -> Poll<()> {
     self.inner().poll_read_ready(cx)
   }
@@ -471,18 +475,35 @@ pub fn op_readable_stream_resource_write_buf(
   }
 }
 
-#[op2(async)]
+/// Write to the channel synchronously, returning 0 if the channel was closed, 1 if we wrote
+/// successfully, 2 if the channel was full and we need to block.
+#[op2]
+pub fn op_readable_stream_resource_write_sync(
+  sender: *const c_void,
+  #[buffer] buffer: JsBuffer,
+) -> u32 {
+  let sender = get_sender(sender);
+  if sender.can_write() {
+    if sender.closed() {
+      0
+    } else {
+      sender.write(buffer.into_parts()).unwrap();
+      1
+    }
+  } else {
+    2
+  }
+}
+
+#[op2(fast)]
 pub fn op_readable_stream_resource_write_error(
   sender: *const c_void,
   #[string] error: String,
-) -> impl Future<Output = bool> {
+) -> bool {
   let sender = get_sender(sender);
-  async move {
-    // We can always write an error, no polling required
-    // TODO(mmastrac): we can remove async from this method
-    sender.write_error(type_error(Cow::Owned(error)));
-    !sender.closed()
-  }
+  // We can always write an error, no polling required
+  sender.write_error(type_error(Cow::Owned(error)));
+  !sender.closed()
 }
 
 #[op2(fast)]