perf(pm): parse version manifests from vec buffers

Author: elrrrrrrr

crates/ruborist/src/service/manifest.rs

@@ -14,19 +14,20 @@ use super::fetch::{
 use super::http::get_client;
 use crate::model::manifest::{CoreVersionManifest, FullManifest};
 
-/// Parse JSON bytes on rayon's CPU thread pool (native) or inline
-/// (wasm32). Keeps the tokio runtime free of `simd_json` work so other
-/// in-flight manifest fetches keep driving network IO while this one
-/// parses.
-pub(crate) async fn parse_json_off_runtime<T>(bytes: Bytes) -> Result<T, anyhow::Error>
+/// Parse a JSON buffer on rayon's CPU thread pool (native) or inline
+/// (wasm32). The buffer is consumed because `simd_json` mutates it in-place.
+/// Keeps the tokio runtime free of `simd_json` work so other in-flight
+/// manifest fetches keep driving network IO while this one parses.
+pub(crate) async fn parse_json_vec_off_runtime<T>(
+    mut parse_buf: Vec<u8>,
+) -> Result<T, anyhow::Error>
 where
     T: serde::de::DeserializeOwned + Send + 'static,
 {
     #[cfg(not(target_arch = "wasm32"))]
     {
         let (tx, rx) = tokio::sync::oneshot::channel();
         rayon::spawn(move || {
-            let mut parse_buf = bytes.to_vec();
             let result = simd_json::serde::from_slice::<T>(&mut parse_buf)
                 .map_err(|e| anyhow!("JSON parse error: {e}"));
             let _ = tx.send(result);
@@ -36,7 +37,6 @@ where
     }
     #[cfg(target_arch = "wasm32")]
     {
-        let mut parse_buf = bytes.to_vec();
         simd_json::serde::from_slice::<T>(&mut parse_buf)
             .map_err(|e| anyhow!("JSON parse error: {e}"))
     }
@@ -71,7 +71,8 @@ pub(crate) async fn parse_full_manifest_off_runtime(
     }
     #[cfg(target_arch = "wasm32")]
     {
-        let mut manifest: FullManifest = parse_json_off_runtime(raw_bytes.clone()).await?;
+        let mut manifest: FullManifest =
+            parse_json_vec_off_runtime(raw_bytes.clone().to_vec()).await?;
         manifest.raw = raw_bytes;
         Ok(manifest)
     }
@@ -220,6 +221,28 @@ pub async fn fetch_full_manifest_fresh(
     }
 }
 
+#[cfg(not(target_arch = "wasm32"))]
+async fn read_body_vec(mut response: reqwest::Response) -> Result<Vec<u8>, FetchError> {
+    let capacity = response
+        .content_length()
+        .and_then(|len| usize::try_from(len).ok())
+        .unwrap_or(0);
+    let mut body = Vec::with_capacity(capacity);
+    while let Some(chunk) = response.chunk().await.map_err(classify_reqwest_error)? {
+        body.extend_from_slice(&chunk);
+    }
+    Ok(body)
+}
+
+#[cfg(target_arch = "wasm32")]
+async fn read_body_vec(response: reqwest::Response) -> Result<Vec<u8>, FetchError> {
+    response
+        .bytes()
+        .await
+        .map(|bytes| bytes.to_vec())
+        .map_err(classify_reqwest_error)
+}
+
 /// Options for fetching a version manifest.
 pub struct FetchVersionManifestOptions<'a> {
     pub registry_url: &'a str,
@@ -230,6 +253,17 @@ pub struct FetchVersionManifestOptions<'a> {
 
 /// Fetch version manifest bytes with retry, without parsing.
 pub async fn fetch_version_manifest_bytes(opts: FetchVersionManifestOptions<'_>) -> Result<Bytes> {
+    fetch_version_manifest_vec(opts).await.map(Bytes::from)
+}
+
+/// Fetch version manifest into a mutable parse buffer with retry.
+///
+/// Unlike full manifests, exact-version manifests do not need to keep raw
+/// response bytes for later extraction. Reading directly into `Vec<u8>` avoids
+/// the hot-path `Bytes -> Vec` copy before `simd_json` parses in place.
+pub(crate) async fn fetch_version_manifest_vec(
+    opts: FetchVersionManifestOptions<'_>,
+) -> Result<Vec<u8>> {
     let url = format!("{}/{}/{}", opts.registry_url, opts.name, opts.spec);
 
     let accept = match opts.format {
@@ -251,7 +285,7 @@ pub async fn fetch_version_manifest_bytes(opts: FetchVersionManifestOptions<'_>)
                     .map_err(classify_reqwest_error)?;
 
                 if response.status().is_success() {
-                    response.bytes().await.map_err(classify_reqwest_error)
+                    read_body_vec(response).await
                 } else {
                     Err(classify_status(response.status(), &url))
                 }
@@ -271,6 +305,36 @@ pub async fn fetch_version_manifest_bytes(opts: FetchVersionManifestOptions<'_>)
 pub async fn fetch_version_manifest(
     opts: FetchVersionManifestOptions<'_>,
 ) -> Result<CoreVersionManifest> {
-    let bytes = fetch_version_manifest_bytes(opts).await?;
-    parse_json_off_runtime::<CoreVersionManifest>(bytes).await
+    let bytes = fetch_version_manifest_vec(opts).await?;
+    parse_json_vec_off_runtime::<CoreVersionManifest>(bytes).await
+}
+
+#[cfg(test)]
+mod tests {
+    use serde::Deserialize;
+
+    use super::*;
+
+    #[derive(Debug, Deserialize, PartialEq)]
+    struct TinyManifest {
+        name: String,
+        version: String,
+    }
+
+    #[tokio::test]
+    async fn parse_json_vec_off_runtime_consumes_mutable_buffer() {
+        let parsed = parse_json_vec_off_runtime::<TinyManifest>(
+            br#"{"name":"demo","version":"1.0.0"}"#.to_vec(),
+        )
+        .await
+        .unwrap();
+
+        assert_eq!(
+            parsed,
+            TinyManifest {
+                name: "demo".to_string(),
+                version: "1.0.0".to_string(),
+            }
+        );
+    }
 }