gpui_web: Stream Fetch response bodies (#62333)

Adds streaming support to the web `FetchHttpClient`: response bodies are
now read incrementally via the Fetch `ReadableStream` API instead of
being buffered whole with `arrayBuffer()`, enabling streaming responses
(e.g. LLM completions) on web. A bounded channel propagates
backpressure, and dropping the body cancels the browser-side stream.

Release Notes:
- N/A

crates/gpui_web/Cargo.toml

@@ -61,6 +61,9 @@ web-sys = { version = "0.3", features = [
     "MouseEvent",
     "Navigator",
     "PointerEvent",
+    "ReadableStream",
+    "ReadableStreamDefaultReader",
+    "ReadableStreamReadResult",
     "ResizeObserver",
     "ResizeObserverBoxOptions",
     "ResizeObserverEntry",

crates/gpui_web/src/http_client.rs

@@ -1,8 +1,16 @@
 use crate::WebDispatcher;
 use anyhow::{Context as _, anyhow};
-use futures::{AsyncReadExt as _, channel::oneshot};
+use futures::{
+    AsyncRead, AsyncReadExt as _, FutureExt as _, SinkExt as _, TryStreamExt as _,
+    channel::{mpsc, oneshot},
+};
 use http_client::{AsyncBody, HttpClient, RedirectPolicy};
-use std::sync::Arc;
+use std::{
+    io,
+    pin::Pin,
+    sync::Arc,
+    task::{Context, Poll},
+};
 use wasm_bindgen::JsCast as _;
 use wasm_bindgen::prelude::*;
 
@@ -172,26 +180,24 @@ async fn fetch(
         }
     }
 
-    // The entire response body is eagerly buffered into memory via
-    // `arrayBuffer()`. The Fetch API does not expose a synchronous
-    // streaming interface; streaming would require `ReadableStream`
-    // interop which is significantly more complex.
-    let body_promise = web_response
-        .array_buffer()
-        .map_err(|error| anyhow!("failed to initiate response body read: {error:?}"))?;
-    let body_value = wasm_bindgen_futures::JsFuture::from(body_promise)
-        .await
-        .map_err(|error| anyhow!("failed to read response body: {error:?}"))?;
-    let array_buffer: js_sys::ArrayBuffer = body_value
-        .dyn_into()
-        .map_err(|error| anyhow!("response body is not an ArrayBuffer: {error:?}"))?;
-    let response_bytes = js_sys::Uint8Array::new(&array_buffer).to_vec();
+    let body = match web_response.body() {
+        Some(stream) => {
+            let reader = stream
+                .get_reader()
+                .dyn_into::<web_sys::ReadableStreamDefaultReader>()
+                .map_err(|error| {
+                    anyhow!("response body reader has an unexpected type: {error:?}")
+                })?;
+            AsyncBody::from_reader(ReadableStreamBody::new(reader))
+        }
+        None => AsyncBody::empty(),
+    };
 
-    builder
-        .body(AsyncBody::from(response_bytes))
-        .map_err(|error| anyhow!(error))
+    builder.body(body).map_err(|error| anyhow!(error))
 }
 
+// Request bodies are buffered into memory because streaming uploads require
+// half-duplex Fetch support that browsers largely don't ship yet.
 async fn read_body_to_bytes(mut body: AsyncBody) -> anyhow::Result<Option<Vec<u8>>> {
     let mut buffer = Vec::new();
     body.read_to_end(&mut buffer).await?;
@@ -201,3 +207,103 @@ async fn read_body_to_bytes(mut body: AsyncBody) -> anyhow::Result<Option<Vec<u8
         Ok(Some(buffer))
     }
 }
+
+const RESPONSE_BODY_CHANNEL_CAPACITY: usize = 8;
+
+struct ReadableStreamBody {
+    chunks: futures::stream::IntoAsyncRead<mpsc::Receiver<io::Result<Vec<u8>>>>,
+    // Dropping this sender resolves the pump's cancellation future, which
+    // cancels the browser-side `ReadableStream`.
+    _cancellation: oneshot::Sender<()>,
+}
+
+impl ReadableStreamBody {
+    fn new(reader: web_sys::ReadableStreamDefaultReader) -> Self {
+        let (chunks_sender, chunks_receiver) = mpsc::channel(RESPONSE_BODY_CHANNEL_CAPACITY);
+        let (cancellation, cancellation_receiver) = oneshot::channel();
+        wasm_bindgen_futures::spawn_local(pump_response_body(
+            reader,
+            chunks_sender,
+            cancellation_receiver,
+        ));
+        Self {
+            chunks: chunks_receiver.into_async_read(),
+            _cancellation: cancellation,
+        }
+    }
+}
+
+impl AsyncRead for ReadableStreamBody {
+    fn poll_read(
+        mut self: Pin<&mut Self>,
+        cx: &mut Context<'_>,
+        buffer: &mut [u8],
+    ) -> Poll<io::Result<usize>> {
+        Pin::new(&mut self.chunks).poll_read(cx, buffer)
+    }
+}
+
+async fn pump_response_body(
+    reader: web_sys::ReadableStreamDefaultReader,
+    mut chunks: mpsc::Sender<io::Result<Vec<u8>>>,
+    cancellation: oneshot::Receiver<()>,
+) {
+    let cancellation = cancellation.fuse();
+    futures::pin_mut!(cancellation);
+
+    loop {
+        let read = wasm_bindgen_futures::JsFuture::from(reader.read()).fuse();
+        futures::pin_mut!(read);
+        let result = futures::select_biased! {
+            _ = cancellation => {
+                cancel_reader(&reader).await;
+                return;
+            }
+            result = read => result,
+        };
+
+        let chunk = result
+            .map_err(|error| io::Error::other(format!("response stream failed: {error:?}")))
+            .and_then(response_chunk);
+        match chunk {
+            Ok(Some(chunk)) => {
+                if chunks.send(Ok(chunk)).await.is_err() {
+                    cancel_reader(&reader).await;
+                    return;
+                }
+            }
+            Ok(None) => return,
+            Err(error) => {
+                if chunks.send(Err(error)).await.is_err() {
+                    log::debug!("response body receiver was dropped after a stream error");
+                }
+                return;
+            }
+        }
+    }
+}
+
+fn response_chunk(result: JsValue) -> io::Result<Option<Vec<u8>>> {
+    // `ReadableStreamReadResult` is a dictionary type, so there is no runtime
+    // class to check against; `unchecked_into` is the only available cast.
+    let result: web_sys::ReadableStreamReadResult = result.unchecked_into();
+    if result.get_done().unwrap_or(false) {
+        return Ok(None);
+    }
+
+    result
+        .get_value()
+        .dyn_into::<js_sys::Uint8Array>()
+        .map(|bytes| Some(bytes.to_vec()))
+        .map_err(|value| {
+            io::Error::other(format!(
+                "response stream yielded a non-byte chunk: {value:?}"
+            ))
+        })
+}
+
+async fn cancel_reader(reader: &web_sys::ReadableStreamDefaultReader) {
+    if let Err(error) = wasm_bindgen_futures::JsFuture::from(reader.cancel()).await {
+        log::debug!("failed to cancel response body reader: {error:?}");
+    }
+}