experiment(pm): install hot-path streaming triplet (TTY-gate + FuturesUnordered + zero-copy)

crates/pm/src/service/install.rs

@@ -1,6 +1,7 @@
 use crate::util::cli_enum::ScriptPolicy;
 use anyhow::Context;
 use anyhow::Result;
+use futures::stream::{FuturesUnordered, StreamExt};
 use std::collections::{HashMap, HashSet};
 use std::path::Path;
 use std::time::Instant;
@@ -21,7 +22,8 @@ use crate::util::downloader::download_stats;
 use crate::util::json::load_package_lock_json_from_path;
 use crate::util::linker::link;
 use crate::util::logger::{
-    PROGRESS_BAR, finish_progress_bar, log_progress, print_install_counts, start_progress_bar,
+    finish_progress_bar, log_progress, print_install_counts, progress_inc, progress_set_length,
+    start_progress_bar,
 };
 use utoo_ruborist::compat::{is_cpu_compatible, is_os_compatible};
 
@@ -83,13 +85,20 @@ pub async fn install_packages(
     depths.sort_unstable();
 
     for depth in depths.iter() {
-        let mut clone_tasks: Vec<tokio::task::JoinHandle<Result<()>>> = Vec::new();
+        // Stream task completion within a layer instead of joining serially.
+        // The previous `for task in clone_tasks { task.await?? }` walks tasks
+        // in spawn order — slow tasks block the pop of already-finished
+        // ones, leaving idle latency between layers on deep trees. With
+        // FuturesUnordered we drain in completion order, propagating the
+        // first error early.
+        let mut clone_tasks: FuturesUnordered<tokio::task::JoinHandle<Result<()>>> =
+            FuturesUnordered::new();
 
         if let Some(packages) = groups.get(depth) {
             for (path, package) in packages.iter() {
                 // Skip packages based on omit config
                 if should_omit_package(package, omit) {
-                    PROGRESS_BAR.inc(1);
+                    progress_inc(1);
                     continue;
                 }
                 let path = path.clone();
@@ -108,28 +117,28 @@ pub async fn install_packages(
                     if package.link.is_some() {
                         let link_name = extract_package_name(&path);
                         if link_name.is_empty() {
-                            PROGRESS_BAR.inc(1);
+                            progress_inc(1);
                             continue;
                         }
                         link(Path::new(&resolved), Path::new(&path))
                             .await
                             .with_context(|| format!("Link failed: {resolved} -> {path}"))?;
-                        PROGRESS_BAR.inc(1);
+                        progress_inc(1);
                         continue;
                     }
 
                     // skip when cpu or os is not compatible
                     if let Some(ref cpu) = package.cpu
                         && !is_cpu_compatible(cpu)
                     {
-                        PROGRESS_BAR.inc(1);
+                        progress_inc(1);
                         continue;
                     }
 
                     if let Some(ref os) = package.os
                         && !is_os_compatible(os)
                     {
-                        PROGRESS_BAR.inc(1);
+                        progress_inc(1);
                         continue;
                     }
 
@@ -153,24 +162,24 @@ pub async fn install_packages(
                                 tracing::warn!(
                                     "Optional dependency {name} failed (ignored): {e:#}"
                                 );
-                                PROGRESS_BAR.inc(1);
+                                progress_inc(1);
                                 return Ok(());
                             }
                             return Err(e);
                         }
-                        PROGRESS_BAR.inc(1);
+                        progress_inc(1);
                         log_progress(&format!("{name} resolved"));
                         update_package_binary(&target_path, &name).await
                     });
                     clone_tasks.push(task);
                 } else {
-                    PROGRESS_BAR.inc(1);
+                    progress_inc(1);
                 }
             }
         }
 
-        for task in clone_tasks {
-            task.await??;
+        while let Some(joined) = clone_tasks.next().await {
+            joined??;
         }
     }
 
@@ -260,7 +269,7 @@ impl InstallService {
 
         if !package_lock.packages.is_empty() {
             start_progress_bar();
-            PROGRESS_BAR.set_length(package_lock.packages.len() as u64);
+            progress_set_length(package_lock.packages.len() as u64);
         }
 
         let link_start = Instant::now();

crates/pm/src/util/extractor.rs

@@ -41,9 +41,15 @@ fn estimate_uncompressed_size(gzip_data: &[u8]) -> usize {
     }
 }
 
+/// Tarball entry recorded during the parse pass, deferred until parallel
+/// write. `data_offset`/`data_len` index into the decompressed buffer
+/// (lifetime-erased from the buffer for reuse during `par_iter`); avoids
+/// the per-file `Vec<u8>` allocation+copy that `read_to_end` previously
+/// did for every file in the archive.
 struct ExtractedEntry {
     path: PathBuf,
-    content: Vec<u8>,
+    data_offset: usize,
+    data_len: usize,
     mode: u32,
 }
 
@@ -70,7 +76,7 @@ fn extract_tarball_sync(gzip_bytes: Bytes, estimated_size: usize, dest: &Path) -
     use rayon::prelude::*;
     use std::collections::HashSet;
     use std::fs;
-    use std::io::{Cursor, Read, Write};
+    use std::io::Write;
 
     // Decompress gzip using libdeflate
     let mut output = vec![0u8; estimated_size];
@@ -90,49 +96,11 @@ fn extract_tarball_sync(gzip_bytes: Bytes, estimated_size: usize, dest: &Path) -
     };
     output.truncate(actual_size);
 
-    // Parse tar entries
-    let cursor = Cursor::new(&output[..]);
-    let mut archive = tar::Archive::new(cursor);
-    let mut entries = Vec::new();
-
-    for entry_result in archive.entries()? {
-        let mut entry = entry_result.with_context(|| "Failed to read tar entry")?;
-        let path = entry
-            .path()
-            .with_context(|| "Failed to get entry path")?
-            .into_owned();
-
-        // Guard against path traversal (Tar Slip): reject absolute paths
-        // and entries containing ".." components.
-        if path.is_absolute()
-            || path
-                .components()
-                .any(|c| matches!(c, std::path::Component::ParentDir))
-        {
-            tracing::warn!("Skipping tar entry with unsafe path: {}", path.display());
-            continue;
-        }
-
-        let full_path = dest.join(&path);
-
-        if !entry.header().entry_type().is_dir() {
-            let mut content = Vec::new();
-            entry
-                .read_to_end(&mut content)
-                .with_context(|| format!("Failed to read tar entry: {}", path.display()))?;
-
-            // Normalize to npm/pnpm convention: 0o755 if any exec bit is set,
-            // else 0o644. Preserving raw tar modes (e.g. 0o640 in google-protobuf)
-            // breaks world-readability in containers and multi-user setups.
-            let raw_mode = entry.header().mode().unwrap_or(0o644);
-            let mode = if raw_mode & 0o111 != 0 { 0o755 } else { 0o644 };
-            entries.push(ExtractedEntry {
-                path: full_path,
-                content,
-                mode,
-            });
-        }
-    }
+    // Parse tar entries — record offsets into `output` instead of copying
+    // each file's bytes. The buffer outlives the parallel write phase and
+    // is read-only after the parse loop, so concurrent slice reads are
+    // safe.
+    let entries = parse_tar_entries(&output, dest)?;
 
     // Collect every ancestor directory (up to `dest`), then create them
     // shallowest-first so a single mkdir() per dir is sufficient.
@@ -152,11 +120,14 @@ fn extract_tarball_sync(gzip_bytes: Bytes, estimated_size: usize, dest: &Path) -
         fs::create_dir(dir).ok();
     }
 
-    // Write files in parallel using rayon
+    // Write files in parallel using rayon. Each task slices into the
+    // shared `output` buffer (immutable borrow) — no per-file copy.
+    let buf: &[u8] = &output;
     entries.par_iter().try_for_each(|entry| -> Result<()> {
         let mut file = fs::File::create(&entry.path)
             .with_context(|| format!("Failed to create: {}", entry.path.display()))?;
-        file.write_all(&entry.content)
+        let data = &buf[entry.data_offset..entry.data_offset + entry.data_len];
+        file.write_all(data)
             .with_context(|| format!("Failed to write: {}", entry.path.display()))?;
 
         // Skip chmod for 0o644 (most files) — File::create() already produces
@@ -181,3 +152,69 @@ fn extract_tarball_sync(gzip_bytes: Bytes, estimated_size: usize, dest: &Path) -
 
     Ok(())
 }
+
+/// Walk the tar stream and collect file entries as `(path, offset, len, mode)`
+/// tuples into the decompressed `buf`. Skips directories, ignores any
+/// path-traversal entries.
+fn parse_tar_entries(buf: &[u8], dest: &Path) -> Result<Vec<ExtractedEntry>> {
+    use std::io::Cursor;
+
+    let cursor = Cursor::new(buf);
+    let mut archive = tar::Archive::new(cursor);
+    let mut entries = Vec::new();
+
+    for entry_result in archive.entries()? {
+        let entry = entry_result.with_context(|| "Failed to read tar entry")?;
+        let path = entry
+            .path()
+            .with_context(|| "Failed to get entry path")?
+            .into_owned();
+
+        // Guard against path traversal (Tar Slip): reject absolute paths
+        // and entries containing ".." components.
+        if path.is_absolute()
+            || path
+                .components()
+                .any(|c| matches!(c, std::path::Component::ParentDir))
+        {
+            tracing::warn!("Skipping tar entry with unsafe path: {}", path.display());
+            continue;
+        }
+
+        if entry.header().entry_type().is_dir() {
+            continue;
+        }
+
+        // tar::Entry exposes the byte offset and size inside the parsed
+        // stream — that's the same `buf` we'll later slice in `par_iter`.
+        let data_offset = entry.raw_file_position() as usize;
+        let data_len = entry.size() as usize;
+        if data_offset
+            .checked_add(data_len)
+            .is_none_or(|end| end > buf.len())
+        {
+            anyhow::bail!(
+                "Tar entry {} extends past decompressed buffer ({}..+{} > {})",
+                path.display(),
+                data_offset,
+                data_len,
+                buf.len()
+            );
+        }
+
+        let full_path = dest.join(&path);
+        // Normalize to npm/pnpm convention: 0o755 if any exec bit is set,
+        // else 0o644. Preserving raw tar modes (e.g. 0o640 in google-protobuf)
+        // breaks world-readability in containers and multi-user setups.
+        let raw_mode = entry.header().mode().unwrap_or(0o644);
+        let mode = if raw_mode & 0o111 != 0 { 0o755 } else { 0o644 };
+        entries.push(ExtractedEntry {
+            path: full_path,
+            data_offset,
+            data_len,
+            mode,
+        });
+    }
+
+    Ok(entries)
+}

crates/pm/src/util/logger.rs

@@ -161,6 +161,27 @@ pub fn log_progress(text: &str) {
     PROGRESS_BAR.set_message(text.to_string());
 }
 
+/// TTY-gated `PROGRESS_BAR.inc(1)`. Non-TTY callers pay no Mutex cost —
+/// indicatif's `inc` always takes the internal `Mutex<ProgressState>`
+/// write-lock even with a hidden draw target, which contends across the
+/// install loop's spawned tasks.
+#[inline]
+pub fn progress_inc(n: u64) {
+    if !*IS_TTY {
+        return;
+    }
+    PROGRESS_BAR.inc(n);
+}
+
+/// TTY-gated `PROGRESS_BAR.set_length`. Same rationale as [`progress_inc`].
+#[inline]
+pub fn progress_set_length(len: u64) {
+    if !*IS_TTY {
+        return;
+    }
+    PROGRESS_BAR.set_length(len);
+}
+
 // Global timer for log_time/log_time_end
 static START_TIME: OnceCell<Instant> = OnceCell::new();