diff --git a/Cargo.lock b/Cargo.lock index b3216e04a0..d6278c391d 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -192,19 +192,6 @@ dependencies = [ "syn 2.0.106", ] -[[package]] -name = "async-compression" -version = "0.3.15" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "942c7cd7ae39e91bde4820d74132e9862e62c2f386c3aa90ccf55949f5bad63a" -dependencies = [ - "flate2", - "futures-core", - "memchr", - "pin-project-lite", - "tokio", -] - [[package]] name = "async-compression" version = "0.4.32" @@ -674,9 +661,9 @@ checksum = "8f1fe948ff07f4bd06c30984e69f5b4899c516a3ef74f34df92a2df2ab535495" [[package]] name = "bytes" -version = "1.10.1" +version = "1.11.1" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "d71b6127be86fdcfddb610f7182ac57211d4b18a3e9c82eb2d17662f2227ad6a" +checksum = "1e748733b7cbc798e1434b6ac524f0c1ff2ab456fe201501e6497c8417a4fc33" dependencies = [ "serde", ] @@ -3431,6 +3418,24 @@ version = "0.2.176" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "58f929b4d672ea937a23a1ab494143d968337a5f47e56d0815df1e0890ddf174" +[[package]] +name = "libdeflate-sys" +version = "1.25.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "23bd6304ebf75390d8a99b88bdf2a266f62647838140cb64af8e6702f6e3fddc" +dependencies = [ + "cc", +] + +[[package]] +name = "libdeflater" +version = "1.25.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d5d4880e6d634d3d029d65fa016038e788cc728a17b782684726fb34ee140caf" +dependencies = [ + "libdeflate-sys", +] + [[package]] name = "libloading" version = "0.8.9" @@ -3459,7 +3464,7 @@ checksum = "416f7e718bdb06000964960ffa43b4335ad4012ae8b99060261aa4a8088d5ccb" dependencies = [ "bitflags 2.9.4", "libc", - "redox_syscall 0.5.18", + "redox_syscall", ] [[package]] @@ -4545,7 +4550,7 @@ checksum = "2621685985a2ebf1c516881c026032ac7deafcda1a2c9b7850dc81e3dfcb64c1" dependencies = [ "cfg-if", "libc", - "redox_syscall 0.5.18", + "redox_syscall", "smallvec", "windows-link 0.2.1", ] @@ -5255,15 +5260,6 @@ dependencies = [ "crossbeam-utils", ] -[[package]] -name = "redox_syscall" -version = "0.3.5" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "567664f262709473930a4bf9e51bf2ebf3348f2e748ccc50dea20646858f8f29" -dependencies = [ - "bitflags 1.3.2", -] - [[package]] name = "redox_syscall" version = "0.5.18" @@ -5418,7 +5414,7 @@ version = "0.12.24" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "9d0946410b9f7b082a427e4ef5c8ff541a88b357bc6c637c40db3a68ac70a36f" dependencies = [ - "async-compression 0.4.32", + "async-compression", "base64 0.22.1", "bytes", "encoding_rs", @@ -8384,21 +8380,6 @@ dependencies = [ "tokio-util", ] -[[package]] -name = "tokio-tar" -version = "0.3.1" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "9d5714c010ca3e5c27114c1cdeb9d14641ace49874aa5626d7149e47aedace75" -dependencies = [ - "filetime", - "futures-core", - "libc", - "redox_syscall 0.3.5", - "tokio", - "tokio-stream", - "xattr", -] - [[package]] name = "tokio-tungstenite" version = "0.18.0" @@ -9718,8 +9699,8 @@ name = "utoo-pm" version = "0.0.0" dependencies = [ "anyhow", - "async-compression 0.3.15", "atty", + "bytes", "chrono", "clap", "colored 2.2.0", @@ -9732,11 +9713,13 @@ dependencies = [ "glob", "indicatif", "libc", + "libdeflater", "mockito", "once_cell", "openssl", "owo-colors", "petgraph 0.6.5", + "rayon", "regex", "reqwest 0.12.24", "serde", @@ -9747,7 +9730,6 @@ dependencies = [ "tokio", "tokio-fs-ext", "tokio-retry", - "tokio-tar", "tokio-util", "toml", "tracing", diff --git a/crates/pm/Cargo.toml b/crates/pm/Cargo.toml index 90874cac0e..a9e6919ddd 100644 --- a/crates/pm/Cargo.toml +++ b/crates/pm/Cargo.toml @@ -11,8 +11,8 @@ path = "src/main.rs" [dependencies] anyhow = { workspace = true } -async-compression = { version = "0.3", features = ["gzip", "tokio"] } atty = "0.2" +bytes = "1.11.0" chrono = { version = "0.4", features = ["serde"] } clap = { workspace = true } colored = "2.1" @@ -25,9 +25,11 @@ futures = { workspace = true } glob = "0.3.1" indicatif = "0.17.8" libc = "0.2" +libdeflater = "1.25" once_cell = "1.19" owo-colors = { workspace = true } petgraph = "0.6" +rayon = { workspace = true } regex = "1.10" reqwest = { version = "0.12", default-features = false, features = [ "gzip", @@ -44,8 +46,7 @@ tempfile = "3.15.0" term_size = "0.3" tokio-fs-ext = { workspace = true } tokio-retry = "0.3" -tokio-tar = "0.3.1" -tokio-util = { version = "0.7", features = ["io"] } +tokio-util = { version = "0.7" } toml = { workspace = true } tracing = { workspace = true } tracing-appender = "0.2" diff --git a/crates/pm/src/cmd/deps.rs b/crates/pm/src/cmd/deps.rs index 3de5a3d012..25a2025999 100644 --- a/crates/pm/src/cmd/deps.rs +++ b/crates/pm/src/cmd/deps.rs @@ -1,18 +1,16 @@ use anyhow::{Context as _, Result}; use std::path::Path; use utoo_ruborist::lock::PackageLock; -use utoo_ruborist::service::build_deps as ruborist_build_deps; -use crate::helper::fs::Context; use crate::helper::lock::save_package_lock; +use crate::helper::ruborist_context::Context; use crate::service::workspace::WorkspaceService; use crate::util::logger::{finish_progress_bar, start_progress_bar}; pub async fn build_deps(cwd: &Path) -> Result { start_progress_bar(); - let options = Context::build_deps_options(cwd.to_path_buf()).await; - let package_lock = ruborist_build_deps(options).await?; + let package_lock = Context::build_deps(cwd.to_path_buf()).await?; finish_progress_bar("package-lock.json resolved"); diff --git a/crates/pm/src/cmd/install.rs b/crates/pm/src/cmd/install.rs index 8ef67224d7..2220bce65b 100644 --- a/crates/pm/src/cmd/install.rs +++ b/crates/pm/src/cmd/install.rs @@ -1,6 +1,8 @@ use anyhow::Result; +use std::path::Path; use crate::service::install::InstallService; +use crate::util::config::get_omit; use crate::util::save_type::{PackageAction, SaveType}; pub async fn update_packages( @@ -15,13 +17,14 @@ pub async fn update_packages( return Err(anyhow::anyhow!("No package specifications provided")); } - // Dispatch to service - InstallService::update_packages(action, specs, workspace, ignore_scripts, save_type).await + let omit = get_omit(); + InstallService::update_packages(action, specs, workspace, ignore_scripts, save_type, &omit) + .await } -pub async fn install(ignore_scripts: bool, root_path: &std::path::Path) -> Result<()> { - // Dispatch to service - InstallService::install(ignore_scripts, root_path).await +pub async fn install(ignore_scripts: bool, root_path: &Path) -> Result<()> { + let omit = get_omit(); + InstallService::install(ignore_scripts, root_path, &omit).await } pub async fn install_global_package(npm_spec: &str, prefix: Option<&str>) -> Result<()> { diff --git a/crates/pm/src/cmd/view.rs b/crates/pm/src/cmd/view.rs index f0996f0b25..60e3233dcc 100644 --- a/crates/pm/src/cmd/view.rs +++ b/crates/pm/src/cmd/view.rs @@ -6,7 +6,7 @@ use utoo_ruborist::registry::resolve_package; use utoo_ruborist::service::fetch_full_manifest; use utoo_ruborist::util::parse_package_spec; -use crate::helper::fs::Context; +use crate::helper::ruborist_context::Context; use crate::util::config::get_registry; use crate::util::format_print::print_grid; diff --git a/crates/pm/src/helper/lock.rs b/crates/pm/src/helper/lock.rs index f42f2f74d8..8dec35cea0 100644 --- a/crates/pm/src/helper/lock.rs +++ b/crates/pm/src/helper/lock.rs @@ -3,17 +3,16 @@ use serde_json::Value; use std::collections::HashMap; use std::path::{Path, PathBuf}; -use super::fs::Context; +use super::ruborist_context::Context; use crate::helper::workspace::find_workspaces; use crate::util::config::get_legacy_peer_deps; use crate::util::json::{load_package_json_from_path, load_package_lock_json_from_path}; use crate::util::logger::{finish_progress_bar, start_progress_bar}; use crate::util::save_type::{PackageAction, SaveType}; -use crate::util::{cloner::clone_package, downloader::download}; +use crate::util::{cloner::clone_package, downloader::download_to_cache}; use utoo_ruborist::lock::{LockPackage, PackageLock}; use utoo_ruborist::manifest::PackageJson; use utoo_ruborist::registry::resolve_package; -use utoo_ruborist::service::build_deps as ruborist_build_deps; use utoo_ruborist::util::parse_package_spec; use super::workspace::find_workspace_path; @@ -92,7 +91,9 @@ pub async fn ensure_package_lock(root_path: &Path) -> Result { if needs_regenerate { tracing::debug!("Resolving dependencies"); start_progress_bar(); - let package_lock = build_deps_with_download(root_path).await?; + + let package_lock = Context::build_deps(root_path.to_path_buf()).await?; + finish_progress_bar("package-lock.json resolved"); // Write to disk asynchronously in background @@ -236,23 +237,20 @@ pub async fn prepare_global_package_json(npm_spec: &str, prefix: Option<&str>) - .as_ref() .ok_or_else(|| anyhow!("Failed to get tarball URL from manifest"))?; - // Download and extract package - let cache_dir = crate::util::cache::get_cache_dir(); - let cache_path = cache_dir.join(format!("{}/{}", name, resolved.version)); - let cache_flag_path = cache_dir.join(format!("{}/{}/_resolved", name, resolved.version)); - - // Download if not cached - if !crate::fs::try_exists(&cache_flag_path).await? { - tracing::debug!("Downloading {} to {}", tarball_url, cache_path.display()); - download(tarball_url, &cache_path) + // Download and extract package to cache + let cache_path = download_to_cache(&name, &resolved.version, tarball_url) + .await + .ok_or_else(|| anyhow!("Failed to download package {name}"))?; + + // If the package has install scripts, create a flag file + // in linux, we can use hardlink when FICLONE is not supported + // so we need to copy the file to the package directory to avoid effect other packages + if resolved.manifest.has_install_script == Some(true) { + let has_install_script_flag_path = cache_path.join("_hasInstallScript"); + if !crate::fs::try_exists(&has_install_script_flag_path) .await - .context("Failed to download package")?; - - // If the package has install scripts, create a flag file - // in linux, we can use hardlink when FICLONE is not supported - // so we need to copy the file to the package directory to avoid effect other packages - if resolved.manifest.has_install_script == Some(true) { - let has_install_script_flag_path = cache_path.join("_hasInstallScript"); + .unwrap_or(false) + { crate::fs::write(has_install_script_flag_path, "").await?; } } @@ -409,13 +407,6 @@ pub async fn is_pkg_lock_outdated(root_path: &Path) -> Result { Ok(false) } -/// Build dependencies with tgz download. -/// Used by `utoo install` command for faster installation. -async fn build_deps_with_download(cwd: &Path) -> Result { - let options = Context::build_deps_options(cwd.to_path_buf()).await; - ruborist_build_deps(options).await -} - /// Save PackageLock to disk synchronously pub async fn save_package_lock(path: &Path, package_lock: &PackageLock) -> Result<()> { let temp_path = path.join("package-lock.json.tmp"); diff --git a/crates/pm/src/helper/mod.rs b/crates/pm/src/helper/mod.rs index 3ea4452a73..0166076105 100644 --- a/crates/pm/src/helper/mod.rs +++ b/crates/pm/src/helper/mod.rs @@ -1,10 +1,10 @@ pub mod auto_update; pub mod deps; -pub mod fs; pub mod fuzzy_select; pub mod global_bin; pub mod install_runtime; pub mod lock; pub mod package; +pub mod ruborist_context; pub mod tree_builder; pub mod workspace; diff --git a/crates/pm/src/helper/fs.rs b/crates/pm/src/helper/ruborist_context.rs similarity index 58% rename from crates/pm/src/helper/fs.rs rename to crates/pm/src/helper/ruborist_context.rs index 77d4235fba..ab7c369d78 100644 --- a/crates/pm/src/helper/fs.rs +++ b/crates/pm/src/helper/ruborist_context.rs @@ -1,8 +1,9 @@ -//! Tokio-based glob implementation for ruborist's Glob trait. +//! Adapter layer bridging pm's configuration to ruborist's API. use std::path::{Path, PathBuf}; use utoo_ruborist::service::{BuildDepsOptions, Glob, UnifiedRegistry}; +use crate::service::pipeline::{PipelineChannels, PipelineReceiver}; use crate::util::cache::get_cache_dir; use crate::util::config::{get_legacy_peer_deps, get_manifests_concurrency_limit, get_registry}; use crate::util::logger::ProgressReceiver; @@ -27,15 +28,16 @@ impl Glob for TokioGlob { // Type aliases to hide concrete Glob type pub(crate) type GlobImpl = TokioGlob; pub(crate) type Registry = UnifiedRegistry; -pub(crate) type DepsOptions = BuildDepsOptions; - /// Context for ruborist operations. /// Centralizes Glob and configuration to avoid spreading concrete types. pub(crate) struct Context; impl Context { - /// Create BuildDepsOptions with standard configuration. - pub async fn build_deps_options(cwd: PathBuf) -> DepsOptions { + /// Create BuildDepsOptions with a custom event receiver. + pub async fn deps_options( + cwd: PathBuf, + receiver: R, + ) -> BuildDepsOptions { BuildDepsOptions { cwd, registry_url: get_registry(), @@ -43,10 +45,29 @@ impl Context { concurrency: get_manifests_concurrency_limit().await, legacy_peer_deps: get_legacy_peer_deps().await, glob: TokioGlob, - receiver: ProgressReceiver, + receiver, } } + /// Create BuildDepsOptions with PipelineReceiver for concurrent download/clone. + /// Returns (options, channels) where channels are used to start pipeline workers. + pub async fn pipeline_deps_options( + cwd: PathBuf, + ) -> ( + BuildDepsOptions>, + PipelineChannels, + ) { + let (receiver, channels) = PipelineReceiver::new(ProgressReceiver); + let options = Self::deps_options(cwd, receiver).await; + (options, channels) + } + + /// Resolve dependency tree with plain ProgressReceiver. Returns PackageLock. + pub async fn build_deps(cwd: PathBuf) -> anyhow::Result { + let options = Self::deps_options(cwd, ProgressReceiver).await; + utoo_ruborist::service::build_deps(options).await + } + /// Create a UnifiedRegistry with standard configuration. pub fn registry() -> Registry { UnifiedRegistry::builder() diff --git a/crates/pm/src/helper/workspace.rs b/crates/pm/src/helper/workspace.rs index 3675448b1c..3a57430b93 100644 --- a/crates/pm/src/helper/workspace.rs +++ b/crates/pm/src/helper/workspace.rs @@ -7,7 +7,7 @@ use anyhow::{Context, Result}; use std::env; use std::path::{Path, PathBuf}; -use super::fs::Context as FsContext; +use super::ruborist_context::Context as FsContext; use utoo_ruborist::manifest::PackageJson; use utoo_ruborist::resolver::workspace::WorkspaceDiscovery; diff --git a/crates/pm/src/main.rs b/crates/pm/src/main.rs index eefd308f2b..8f87c76b23 100644 --- a/crates/pm/src/main.rs +++ b/crates/pm/src/main.rs @@ -251,19 +251,20 @@ enum Commands { } fn main() { - let parallelism = std::thread::available_parallelism() + // Windows default thread stack is 1MB, insufficient for libdeflater + tar + rayon work-stealing. + #[cfg(target_os = "windows")] + rayon::ThreadPoolBuilder::new() + .stack_size(8 * 1024 * 1024) + .build_global() + .ok(); + + let worker_threads = std::thread::available_parallelism() .map(|n| n.get()) .unwrap_or(4); - // Minimum 12 blocking threads for CI environments with fewer cores - let blocking_threads = (parallelism * 2).max(12); let result = tokio::runtime::Builder::new_multi_thread() .enable_all() - .worker_threads(parallelism) - .max_blocking_threads(blocking_threads) - .thread_name("utoo-worker") - .on_thread_stop(|| {}) - .on_thread_park(|| {}) + .worker_threads(worker_threads) .build() .expect("failed to build tokio runtime") .block_on(async_main()); diff --git a/crates/pm/src/service/clean.rs b/crates/pm/src/service/clean.rs new file mode 100644 index 0000000000..5e2d1affdf --- /dev/null +++ b/crates/pm/src/service/clean.rs @@ -0,0 +1,227 @@ +use anyhow::{Context, Result}; +use glob::glob; +use std::collections::{HashMap, HashSet}; +use std::path::Path; + +use crate::helper::lock::{Package, path_to_pkg_name}; +use crate::helper::workspace; + +/// Remove a symlink with platform-specific handling. +/// +/// On Windows, directory symlinks require `remove_dir` instead of `remove_file`. +async fn remove_symlink(path: &Path) -> Result<(), std::io::Error> { + #[cfg(windows)] + { + use std::os::windows::fs::MetadataExt; + + let metadata = crate::fs::symlink_metadata(path).await?; + if !metadata.file_type().is_symlink() { + return Ok(()); + } + + const FILE_ATTRIBUTE_DIRECTORY: u32 = 0x10; + if (metadata.file_attributes() & FILE_ATTRIBUTE_DIRECTORY) != 0 { + crate::fs::remove_dir(path).await + } else { + crate::fs::remove_file(path).await + } + } + + #[cfg(not(windows))] + { + crate::fs::remove_file(path).await + } +} + +/// Check if a directory name matches legacy npminstall pattern (e.g. `_pkg@1.0.0@2.0.0`). +fn is_legacy_npminstall(name: &str) -> bool { + let at_count = name.matches('@').count(); + name.starts_with('_') && (at_count == 2 || at_count == 4) +} + +/// Remove symlinks inside a scoped package directory (e.g. `node_modules/@scope/`). +async fn remove_scoped_symlinks(scope_dir: &Path) -> Result<()> { + let Ok(mut entries) = crate::fs::read_dir(scope_dir).await else { + return Ok(()); + }; + while let Ok(Some(entry)) = entries.next_entry().await { + let path = entry.path(); + if crate::fs::symlink_metadata(&path) + .await? + .file_type() + .is_symlink() + { + tracing::debug!("Removing scoped symlink: {}", path.display()); + remove_symlink(&path).await.ok(); + } + } + Ok(()) +} + +/// Remove stale symlinks and legacy npminstall artifacts from a node_modules directory. +async fn remove_stale_entries(node_modules: &Path) -> Result<()> { + let Ok(mut entries) = crate::fs::read_dir(node_modules).await else { + return Ok(()); + }; + + while let Ok(Some(entry)) = entries.next_entry().await { + let path = entry.path(); + let metadata = crate::fs::symlink_metadata(&path).await?; + + if metadata.file_type().is_symlink() { + tracing::debug!("Removing symlink: {}", path.display()); + remove_symlink(&path).await.ok(); + continue; + } + + if !metadata.is_dir() { + continue; + } + + let name = entry.file_name(); + let name = name.to_string_lossy(); + if name.starts_with('@') { + remove_scoped_symlinks(&path).await?; + } else if is_legacy_npminstall(&name) { + tracing::debug!("Removing legacy package: {}", path.display()); + crate::fs::remove_dir_all(&path).await.ok(); + } + } + + Ok(()) +} + +/// Remove packages not in `valid_packages`, recursing into nested node_modules. +async fn remove_unused_packages( + node_modules: &Path, + cwd: &Path, + valid_packages: &HashSet, +) -> Result<()> { + let mut stack = vec![node_modules.to_path_buf()]; + + while let Some(current_nm) = stack.pop() { + let patterns = [ + current_nm.join("*/package.json"), + current_nm.join("@*/*/package.json"), + ]; + for pattern in &patterns { + let pattern_str = pattern.to_string_lossy(); + let entries = glob(&pattern_str) + .with_context(|| format!("Glob failed for pattern: {pattern_str}"))?; + + for entry in entries { + let pkg_json = entry.with_context(|| format!("Glob entry error: {pattern_str}"))?; + let pkg_dir = pkg_json + .parent() + .context("Failed to get parent of package.json")?; + let rel_path = pkg_dir.strip_prefix(cwd).with_context(|| { + format!( + "Failed to strip prefix {} from {}", + cwd.display(), + pkg_dir.display() + ) + })?; + + if let Some(pkg_name) = path_to_pkg_name(&pkg_dir.to_string_lossy()) + && !valid_packages.contains(rel_path.to_string_lossy().as_ref()) + { + tracing::debug!("Cleaning unused package: {pkg_name}"); + crate::fs::remove_dir_all(pkg_dir).await.ok(); + } + + let nested_nm = pkg_dir.join("node_modules"); + if crate::fs::try_exists(&nested_nm).await? { + stack.push(nested_nm); + } + } + } + } + + Ok(()) +} + +/// Clean unused dependencies across all workspace node_modules directories. +pub async fn clean_deps(groups: &HashMap>, cwd: &Path) -> Result<()> { + let valid_packages: HashSet = groups + .values() + .flat_map(|pkgs| pkgs.iter().map(|(path, _)| path.clone())) + .collect(); + + tracing::debug!("Valid packages: {valid_packages:?}"); + + let mut nm_dirs = vec![cwd.join("node_modules")]; + for (_, ws_path, _) in workspace::find_workspaces(cwd).await? { + let ws_nm = ws_path.join("node_modules"); + if crate::fs::try_exists(&ws_nm).await? { + tracing::debug!("Adding workspace node_modules: {}", ws_nm.display()); + nm_dirs.push(ws_nm); + } + } + + for nm_dir in &nm_dirs { + remove_stale_entries(nm_dir).await?; + remove_unused_packages(nm_dir, cwd, &valid_packages).await?; + } + + Ok(()) +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::fs; + use tempfile::TempDir; + + #[tokio::test] + async fn test_remove_symlink() -> Result<()> { + let temp_dir = TempDir::new()?; + let target = temp_dir.path().join("utoo-cli"); + let link = temp_dir.path().join("symlink"); + + fs::create_dir(&target).await?; + + #[cfg(unix)] + std::os::unix::fs::symlink(&target, &link)?; + #[cfg(windows)] + std::os::windows::fs::symlink_dir(&target, &link)?; + + assert!(fs::symlink_metadata(&link).await?.file_type().is_symlink()); + + remove_symlink(&link).await?; + + assert!(!link.exists()); + assert!(target.exists()); + Ok(()) + } + + #[tokio::test] + async fn test_remove_scoped_symlinks() -> Result<()> { + let temp_dir = TempDir::new()?; + let scope_dir = temp_dir.path().join("@utoo"); + fs::create_dir(&scope_dir).await?; + + let target = temp_dir.path().join("utoo-cli"); + let link = scope_dir.join("cli"); + fs::create_dir(&target).await?; + + #[cfg(unix)] + std::os::unix::fs::symlink(&target, &link)?; + #[cfg(windows)] + std::os::windows::fs::symlink_dir(&target, &link)?; + + assert!(fs::symlink_metadata(&link).await?.file_type().is_symlink()); + + remove_scoped_symlinks(&scope_dir).await?; + + assert!(!link.exists()); + assert!(target.exists()); + Ok(()) + } + + #[tokio::test] + async fn test_is_legacy_npminstall() { + assert!(is_legacy_npminstall("_utoo-cli@1.0.0@2.0.0")); + assert!(!is_legacy_npminstall("lodash")); + assert!(!is_legacy_npminstall("_lodash@1.0.0")); + } +} diff --git a/crates/pm/src/service/install.rs b/crates/pm/src/service/install.rs index 8588940d9d..d8968ea8c6 100644 --- a/crates/pm/src/service/install.rs +++ b/crates/pm/src/service/install.rs @@ -1,35 +1,24 @@ use anyhow::Context; use anyhow::Result; -use glob::glob; -use std::collections::HashMap; -use std::future::Future; +use std::collections::{HashMap, HashSet}; use std::path::Path; -use std::pin::Pin; -use std::sync::Arc; -use tokio::sync::Semaphore; use crate::helper::global_bin::get_global_bin_dir; use crate::helper::lock::{ - Package, ensure_package_lock, extract_package_name, group_by_depth, path_to_pkg_name, - prepare_global_package_json, update_package_json, + Package, extract_package_name, group_by_depth, prepare_global_package_json, update_package_json, }; -use crate::helper::workspace; use crate::model::package::PackageInfo; use crate::service::rebuild::RebuildService; -use crate::util::cache::get_cache_dir; -use crate::util::cloner::clone_package; -use crate::util::downloader::download; use crate::util::linker::link; use crate::util::logger::{PROGRESS_BAR, finish_progress_bar, log_progress, start_progress_bar}; use crate::util::save_type::{OmitType, PackageAction, SaveType}; use utoo_ruborist::compat::{is_cpu_compatible, is_os_compatible}; use super::binary::update_package_binary; - -static CONCURRENT_LIMIT: usize = 40; +use super::clean::clean_deps; /// Check if a package should be omitted based on omit config -fn should_omit_package(package: &Package, omit: &std::collections::HashSet) -> bool { +fn should_omit_package(package: &Package, omit: &HashSet) -> bool { if omit.is_empty() { return false; } @@ -62,234 +51,30 @@ fn should_omit_package(package: &Package, omit: &std::collections::HashSet Result { - Ok(crate::fs::symlink_metadata(path) - .await? - .file_type() - .is_symlink()) -} - -/// Remove a symlink with proper platform-specific handling -async fn remove_symlink_cross_platform(path: &Path) -> Result<(), std::io::Error> { - #[cfg(windows)] - { - use std::os::windows::fs::MetadataExt; - - // On Windows, we need to check if the symlink points to a directory - let metadata = crate::fs::symlink_metadata(path).await?; - - if !metadata.file_type().is_symlink() { - return Ok(()); - } - - // Check if it's a directory symlink by checking the file attributes - const FILE_ATTRIBUTE_DIRECTORY: u32 = 0x10; - let is_dir_symlink = (metadata.file_attributes() & FILE_ATTRIBUTE_DIRECTORY) != 0; - - if is_dir_symlink { - crate::fs::remove_dir(path).await - } else { - crate::fs::remove_file(path).await - } - } - - #[cfg(not(windows))] - { - crate::fs::remove_file(path).await - } -} - -/// Clean up a single node_modules directory -async fn clean_node_modules_dir( - node_modules: &Path, - cwd: &Path, - valid_packages: &std::collections::HashSet, -) -> Result<()> { - // clean up symlinks for npminstall - if let Ok(mut entries) = crate::fs::read_dir(node_modules).await { - while let Ok(Some(entry)) = entries.next_entry().await { - let path = entry.path(); - if is_symlink_async(&path).await? { - clean_symlink(&path).await?; - } else if path.is_dir() { - clean_directory(&path).await?; - } - } - } - - clean_unused_packages(node_modules, cwd, valid_packages).await?; - - Ok(()) -} - -/// Clean up a symlink -async fn clean_symlink(path: &Path) -> Result<()> { - tracing::debug!("Removing symlink: {}", path.display()); - - if let Err(e) = remove_symlink_cross_platform(path).await { - tracing::debug!("Failed to remove symlink {}: {}", path.display(), e); - } - - Ok(()) -} - -/// Clean up a directory, handling scoped packages and legacy npm install packages -async fn clean_directory(path: &Path) -> Result<()> { - if let Some(file_name) = path.file_name() { - let name = file_name.to_string_lossy(); - if name.starts_with('@') { - clean_scoped_package(path).await?; - } else { - clean_legacy_npminstall_package(path, &name).await?; - } - } - Ok(()) -} - -/// Clean up a scoped package directory -async fn clean_scoped_package(path: &Path) -> Result<()> { - if let Ok(mut scope_entries) = crate::fs::read_dir(path).await { - while let Ok(Some(scope_entry)) = scope_entries.next_entry().await { - let scope_path = scope_entry.path(); - - // Use async metadata check for symlink - if is_symlink_async(&scope_path).await? { - tracing::debug!("Removing scoped symlink: {}", scope_path.display()); - - if let Err(e) = remove_symlink_cross_platform(&scope_path).await { - tracing::debug!( - "Failed to remove scoped symlink {}: {}", - scope_path.display(), - e - ); - } - } - } - } - Ok(()) -} - -/// Clean up a legacy npminstall package -async fn clean_legacy_npminstall_package(path: &Path, name: &str) -> Result<()> { - let at_count = name.matches('@').count(); - if name.starts_with('_') && (at_count == 2 || at_count == 4) { - tracing::debug!("Removing legacy package: {}", path.display()); - if let Err(e) = crate::fs::remove_dir_all(path).await { - tracing::debug!("Failed to remove legacy package {}: {}", path.display(), e); - } - } - Ok(()) -} - -/// Clean up unused packages in the node_modules directory -async fn clean_unused_packages( - node_modules: &Path, - cwd: &Path, - valid_packages: &std::collections::HashSet, -) -> Result<()> { - // Helper function for recursive search - fn find_and_clean<'a>( - node_modules: &'a Path, - cwd: &'a Path, - valid_packages: &'a std::collections::HashSet, - ) -> Pin> + Send + 'a>> { - Box::pin(async move { - let patterns = [ - node_modules.join("*/package.json"), - node_modules.join("@*/*/package.json"), - ]; - for pattern in patterns.iter() { - let pattern_str = pattern.to_string_lossy().to_string(); - for entry in glob(&pattern_str) - .with_context(|| format!("Glob failed for pattern: {pattern_str}"))? - { - let pkg_json_path = entry - .with_context(|| format!("Glob entry error for pattern: {pattern_str}"))?; - let pkg_dir = pkg_json_path - .parent() - .context("Failed to get parent directory of package.json")?; - if let Some(pkg_name) = path_to_pkg_name(&pkg_dir.to_string_lossy()) { - let pkg_path = pkg_dir.strip_prefix(cwd).with_context(|| { - format!( - "Failed to strip prefix {} from {}", - cwd.display(), - pkg_dir.display() - ) - })?; - if !valid_packages.contains(pkg_path.to_string_lossy().as_ref()) { - tracing::debug!("Cleaning unused package: {pkg_name}"); - if let Err(e) = crate::fs::remove_dir_all(pkg_dir).await { - tracing::debug!("Failed to remove {pkg_name}: {e}"); - } - } - } - // Recursively check nested node_modules - let nested_node_modules = pkg_dir.join("node_modules"); - if crate::fs::try_exists(&nested_node_modules).await? { - find_and_clean(&nested_node_modules, cwd, valid_packages).await?; - } - } - } - Ok(()) - }) - } - find_and_clean(node_modules, cwd, valid_packages).await?; - Ok(()) -} - -async fn clean_deps(groups: &HashMap>, cwd: &Path) -> Result<()> { - let mut valid_packages = std::collections::HashSet::new(); - for packages in groups.values() { - for (path, _) in packages { - valid_packages.insert(path.clone()); - } - } - - tracing::debug!("Valid packages: {valid_packages:?}"); - - let mut node_modules_dirs = vec![cwd.join("node_modules")]; - - let workspaces = workspace::find_workspaces(cwd).await?; - for (_, path, _) in workspaces { - let workspace_node_modules = path.join("node_modules"); - if crate::fs::try_exists(&workspace_node_modules).await? { - node_modules_dirs.push(workspace_node_modules.clone()); - tracing::debug!( - "add workspace node_modules: {:?}", - workspace_node_modules.display() - ); - } - } - - // cleanup unused packages in all workspace_members - for node_modules in node_modules_dirs { - clean_node_modules_dir(&node_modules, cwd, &valid_packages).await?; - } - - Ok(()) -} - pub async fn install_packages( - groups: &HashMap>, - cache_dir: &Path, + groups: &HashMap>, cwd: &Path, - semaphore: Arc, + omit: &HashSet, ) -> Result<()> { + use crate::util::cloner::clone_package_once; + // clean unused deps clean_deps(groups, cwd).await?; + // Always process level-by-level to ensure parent directories exist before + // children. Within each level, tasks run concurrently. The pipeline's + // clone_worker may have already cloned some packages — clone_package_once + // deduplicates via CLONE_CACHE so no double work occurs. let mut depths: Vec<_> = groups.keys().cloned().collect(); depths.sort_unstable(); - let omit = crate::util::config::get_omit(); - for depth in depths.iter() { + let mut clone_tasks: Vec>> = Vec::new(); + if let Some(packages) = groups.get(depth) { - let mut tasks: Vec>> = Vec::new(); for (path, package) in packages.iter() { // Skip packages based on omit config - if should_omit_package(package, &omit) { + if should_omit_package(package, omit) { PROGRESS_BAR.inc(1); tracing::debug!("Skipping omitted dependency: {}", path); continue; @@ -338,100 +123,36 @@ pub async fn install_packages( .version .clone() .ok_or_else(|| anyhow::anyhow!("package {name} missing version"))?; - let cache_path = cache_dir.join(format!("{name}/{version}")); - let cache_flag_path = cache_dir.join(format!("{name}/{version}/_resolved")); let cwd_clone = cwd.to_path_buf(); - let should_resolve = !crate::fs::try_exists(&cache_flag_path).await?; - let semaphore = Arc::clone(&semaphore); + let target_path = cwd_clone.join(&path); // Check if this is an optional dependency let is_optional = package.optional == Some(true) || package.dev_optional == Some(true); let task = tokio::spawn(async move { - let _permit = semaphore - .acquire() - .await - .expect("semaphore should not be closed"); - if should_resolve { - tracing::debug!("Downloading {path} to {name}"); - match download(&resolved, &cache_path).await { - Ok(_) => { - log_progress(&format!("{name} downloaded")); - if package.has_install_script.is_some() { - tracing::debug!("{name} has install script"); - let has_install_script_flag_path = - cache_path.join("_hasInstallScript"); - crate::fs::write(has_install_script_flag_path, "").await?; - } - } - Err(e) => { - tracing::debug!( - "Download failed: source={}, target={}, error={}", - resolved, - cache_path.display(), - e - ); - if is_optional { - tracing::warn!( - "Optional dependency {name} download failed (ignored): {e}" - ); - PROGRESS_BAR.inc(1); - return Ok(()); - } - return Err(anyhow::anyhow!("{name} download failed: {e}")); - } - } - } - - tracing::debug!("{name} clone"); - match clone_package(&cache_path, &cwd_clone.join(&path), &name, &version) - .await - { - Ok(_) => { - tracing::debug!("{name} resolved"); + if !clone_package_once(&name, &version, &resolved, &target_path).await { + if is_optional { + tracing::warn!("Optional dependency {name} failed (ignored)"); PROGRESS_BAR.inc(1); - log_progress(&format!("{name} resolved")); - update_package_binary(&cwd_clone.join(&path), &name).await?; - Ok(()) - } - Err(e) => { - if is_optional { - tracing::warn!( - "Optional dependency {name} clone failed (ignored): {e}" - ); - PROGRESS_BAR.inc(1); - return Ok(()); - } - Err(anyhow::anyhow!( - "Copy failed {} to {}: {}", - cache_path.display(), - cwd_clone.join(&path).display(), - e - )) + return Ok(()); } + anyhow::bail!("{name} clone failed"); } + PROGRESS_BAR.inc(1); + log_progress(&format!("{name} resolved")); + update_package_binary(&target_path, &name).await }); - tasks.push(task); + clone_tasks.push(task); } else { PROGRESS_BAR.inc(1); tracing::debug!("{path} no resolved info skipped"); } } + } - for task in tasks { - match task.await { - Ok(Ok(())) => continue, - Ok(Err(e)) => { - tracing::debug!("Task execution error: {e}"); - return Err(anyhow::anyhow!("Error during installation: {e}")); - } - Err(e) => { - tracing::debug!("Task join error: {e}"); - return Err(anyhow::anyhow!("Task execution failed: {e}")); - } - } - } + for task in clone_tasks { + task.await??; } } @@ -447,6 +168,7 @@ impl InstallService { workspace: Option, ignore_scripts: bool, save_type: SaveType, + omit: &HashSet, ) -> Result<()> { tracing::debug!( "update packages: {:?} {:?} {:?} {:?}", @@ -475,36 +197,51 @@ impl InstallService { .await .context("Failed to build package-lock.json")?; - Self::install(ignore_scripts, &root_path) + Self::install(ignore_scripts, &root_path, omit) .await .context("Failed to install packages")?; Ok(()) } - pub async fn install(ignore_scripts: bool, root_path: &Path) -> Result<()> { - // Get PackageLock directly, avoiding redundant disk read/parse operations - let package_lock = ensure_package_lock(root_path).await?; - - let cache_dir = get_cache_dir(); + pub async fn install( + ignore_scripts: bool, + root_path: &Path, + omit: &HashSet, + ) -> Result<()> { + let lock_path = root_path.join("package-lock.json"); + + let (package_lock, pipeline_handles) = + if crate::fs::try_exists(&lock_path).await.unwrap_or(false) { + // Lock exists: install directly, skip manifest resolution + let lock: utoo_ruborist::lock::PackageLock = + crate::util::json::read_json_file(&lock_path).await?; + (lock, None) + } else { + // No lock: full pipeline flow (resolve + concurrent download/clone) + start_progress_bar(); + let result = super::pipeline::resolve_with_pipeline(root_path).await?; + finish_progress_bar("package-lock.json resolved"); + (result.package_lock, Some(result.handles)) + }; let groups = group_by_depth(&package_lock.packages); - let mut depths: Vec<_> = groups.keys().cloned().collect(); - depths.sort_unstable(); if !package_lock.packages.is_empty() { start_progress_bar(); PROGRESS_BAR.set_length(package_lock.packages.len() as u64); } - // Set concurrent limit for package installation - tracing::debug!("Setting concurrent limit to {CONCURRENT_LIMIT}"); - let semaphore = Arc::new(Semaphore::new(CONCURRENT_LIMIT)); - - install_packages(&groups, &cache_dir, root_path, semaphore) + install_packages(&groups, root_path, omit) .await .context("Failed to install packages")?; + // Wait for pipeline workers to complete (if any) + if let Some(handles) = pipeline_handles { + handles.await_completion().await; + super::pipeline::print_pipeline_summary(); + } + finish_progress_bar("node_modules cloned"); RebuildService::rebuild(&package_lock, root_path, ignore_scripts).await?; @@ -519,8 +256,8 @@ impl InstallService { tracing::debug!("Installing global package: {npm_spec}"); - // Install dependencies - Self::install(false, &package_path) + // Install dependencies (global install never omits) + Self::install(false, &package_path, &HashSet::new()) .await .context("Failed to install global package dependencies")?; @@ -550,80 +287,6 @@ impl InstallService { #[cfg(test)] mod tests { use super::*; - use crate::fs; - use tempfile::TempDir; - - #[tokio::test] - async fn test_clean_symlink() -> Result<()> { - let temp_dir = TempDir::new()?; - let target_dir = temp_dir.path().join("utoo-cli"); - let symlink_path = temp_dir.path().join("symlink"); - - // Create target directory - fs::create_dir(&target_dir).await?; - - // Create symlink - #[cfg(unix)] - std::os::unix::fs::symlink(&target_dir, &symlink_path)?; - #[cfg(windows)] - std::os::windows::fs::symlink_dir(&target_dir, &symlink_path)?; - - // Verify symlink exists - assert!(is_symlink_async(&symlink_path).await?); - - // Test cleaning - clean_symlink(&symlink_path).await?; - - // Verify symlink is removed - assert!(!symlink_path.exists()); - assert!(target_dir.exists()); - - Ok(()) - } - - #[tokio::test] - async fn test_clean_scoped_package() -> Result<()> { - let temp_dir = TempDir::new()?; - let scope_dir = temp_dir.path().join("@utoo"); - fs::create_dir(&scope_dir).await?; - - // Create a symlink in the scope directory - let target_dir = temp_dir.path().join("utoo-cli"); - let symlink_path = scope_dir.join("cli"); - fs::create_dir(&target_dir).await?; - - #[cfg(unix)] - std::os::unix::fs::symlink(&target_dir, &symlink_path)?; - #[cfg(windows)] - std::os::windows::fs::symlink_dir(&target_dir, &symlink_path)?; - - // Verify symlink exists - assert!(is_symlink_async(&symlink_path).await?); - - // Test cleaning - clean_scoped_package(&scope_dir).await?; - - // Verify symlink is removed - assert!(!symlink_path.exists()); - assert!(target_dir.exists()); - - Ok(()) - } - - #[tokio::test] - async fn test_clean_legacy_npminstall_package() -> Result<()> { - let temp_dir = TempDir::new()?; - let legacy_dir = temp_dir.path().join("_utoo-cli@1.0.0@2.0.0"); - fs::create_dir(&legacy_dir).await?; - - // Test cleaning - clean_legacy_npminstall_package(&legacy_dir, "_utoo-cli@1.0.0@2.0.0").await?; - - // Verify directory is removed - assert!(!legacy_dir.exists()); - - Ok(()) - } #[test] fn test_extract_package_name_from_path() { diff --git a/crates/pm/src/service/mod.rs b/crates/pm/src/service/mod.rs index 29eae7fc4a..f390a280d9 100644 --- a/crates/pm/src/service/mod.rs +++ b/crates/pm/src/service/mod.rs @@ -1,10 +1,12 @@ pub mod binary; +pub mod clean; pub mod config; pub mod dependency_graph; pub mod execute; pub mod install; pub mod package; pub mod package_management; +pub mod pipeline; pub mod rebuild; pub mod script; pub mod update; diff --git a/crates/pm/src/service/pipeline/mod.rs b/crates/pm/src/service/pipeline/mod.rs new file mode 100644 index 0000000000..ceb001fa5b --- /dev/null +++ b/crates/pm/src/service/pipeline/mod.rs @@ -0,0 +1,52 @@ +//! Pipeline installer for concurrent manifest resolution and tarball downloading. +//! +//! This module implements a pipeline architecture similar to bun's approach: +//! - Manifest fetching and tarball downloading happen concurrently +//! - When a package is resolved, its tarball download starts immediately +//! - Uses global OnceMap to deduplicate requests and share results across phases + +mod receiver; +mod worker; + +pub use receiver::{PipelineChannels, PipelineReceiver}; +pub use worker::PipelineHandles; + +use crate::util::cloner::clone_count; +use crate::util::downloader::download_count; + +/// Print pipeline summary stats. +pub fn print_pipeline_summary() { + tracing::debug!( + "Pipeline stats: downloaded={}, cloned={}", + download_count(), + clone_count(), + ); +} + +/// Result of pipeline-based dependency resolution. +pub struct PipelineResult { + pub package_lock: utoo_ruborist::lock::PackageLock, + pub handles: PipelineHandles, +} + +/// Resolve dependencies with pipeline: concurrent download/clone during resolution. +/// +/// Creates PipelineReceiver via Context, starts workers, runs build_deps, saves lock file. +/// Returns both the lock and worker handles for the caller to await after install. +/// Note: caller is responsible for managing progress bar lifecycle. +pub async fn resolve_with_pipeline(root_path: &std::path::Path) -> anyhow::Result { + use crate::helper::lock::save_package_lock; + use crate::helper::ruborist_context::Context; + + let (options, channels) = Context::pipeline_deps_options(root_path.to_path_buf()).await; + let handles = worker::start_workers(channels, root_path.to_path_buf()); + + let package_lock = utoo_ruborist::service::build_deps(options).await?; + + save_package_lock(root_path, &package_lock).await?; + + Ok(PipelineResult { + package_lock, + handles, + }) +} diff --git a/crates/pm/src/service/pipeline/receiver.rs b/crates/pm/src/service/pipeline/receiver.rs new file mode 100644 index 0000000000..376a5b50c7 --- /dev/null +++ b/crates/pm/src/service/pipeline/receiver.rs @@ -0,0 +1,132 @@ +use std::path::PathBuf; + +use tokio::sync::mpsc; +use utoo_ruborist::progress::{BuildEvent, EventReceiver, PackageTarballInfo}; + +/// Owned version of PackageTarballInfo for channel transmission. +#[derive(Debug, Clone)] +pub struct OwnedPackageInfo { + pub name: String, + pub version: String, + pub tarball_url: Option, +} + +impl From<&PackageTarballInfo<'_>> for OwnedPackageInfo { + fn from(info: &PackageTarballInfo<'_>) -> Self { + Self { + name: info.name.to_string(), + version: info.version.to_string(), + tarball_url: info.tarball_url.map(|s| s.to_string()), + } + } +} + +/// Message for clone worker. +#[derive(Debug)] +pub struct CloneMessage { + pub info: OwnedPackageInfo, + pub path: PathBuf, + /// Parent package path (None for root-level dependencies) + pub parent_path: Option, +} + +/// Pipeline receiver that wraps an inner receiver and forwards events to channels. +/// +/// On `PackageResolved` → sends to download channel. +/// On `PackagePlaced` → sends to clone channel. +/// All events are also forwarded to the inner receiver (e.g. progress bar). +pub struct PipelineReceiver { + download_tx: mpsc::UnboundedSender, + clone_tx: mpsc::UnboundedSender, + inner: R, +} + +/// Channels returned when creating a PipelineReceiver. +pub struct PipelineChannels { + pub download_rx: mpsc::UnboundedReceiver, + pub clone_rx: mpsc::UnboundedReceiver, +} + +impl PipelineReceiver { + /// Create a new pipeline receiver wrapping an inner receiver. + pub fn new(inner: R) -> (Self, PipelineChannels) { + let (download_tx, download_rx) = mpsc::unbounded_channel(); + let (clone_tx, clone_rx) = mpsc::unbounded_channel(); + ( + Self { + download_tx, + clone_tx, + inner, + }, + PipelineChannels { + download_rx, + clone_rx, + }, + ) + } +} + +impl EventReceiver for PipelineReceiver { + fn on_event(&self, event: BuildEvent<'_>) { + // Forward to inner receiver first (for progress bar updates) + self.inner.on_event(event); + + match event { + BuildEvent::PackageResolved(info) + if info.tarball_url.is_some() && info.is_platform_compatible() => + { + let _ = self.download_tx.send(OwnedPackageInfo::from(&info)); + } + BuildEvent::PackagePlaced { + package, + path, + parent_path, + } if package.tarball_url.is_some() && package.is_platform_compatible() => { + let _ = self.clone_tx.send(CloneMessage { + info: OwnedPackageInfo::from(&package), + path: path.to_path_buf(), + parent_path: parent_path.map(|p| p.to_path_buf()), + }); + } + _ => {} + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + use utoo_ruborist::progress::NoopReceiver; + + #[test] + fn test_pipeline_receiver_filters_events() { + let (receiver, mut channels) = PipelineReceiver::new(NoopReceiver); + + // Should forward PackageResolved with tarball_url + receiver.on_event(BuildEvent::PackageResolved(PackageTarballInfo { + name: "react", + version: "18.2.0", + tarball_url: Some("https://registry.npmjs.org/react/-/react-18.2.0.tgz"), + integrity: Some("sha512-xxx"), + os: None, + cpu: None, + })); + + // Should not forward PackageResolved without tarball_url + receiver.on_event(BuildEvent::PackageResolved(PackageTarballInfo { + name: "local-pkg", + version: "1.0.0", + tarball_url: None, + integrity: None, + os: None, + cpu: None, + })); + + // Should not forward other events + receiver.on_event(BuildEvent::PreloadStart { count: 10 }); + + // Only one message should be in the download channel + assert!(channels.download_rx.try_recv().is_ok()); + assert!(channels.download_rx.try_recv().is_err()); + } +} diff --git a/crates/pm/src/service/pipeline/worker.rs b/crates/pm/src/service/pipeline/worker.rs new file mode 100644 index 0000000000..0e57ff0b73 --- /dev/null +++ b/crates/pm/src/service/pipeline/worker.rs @@ -0,0 +1,61 @@ +use std::path::PathBuf; + +use crate::util::cloner::{clone_package_once, wait_clone_if_pending}; +use crate::util::downloader::download_to_cache; + +use super::receiver::PipelineChannels; + +/// Pipeline worker handles for awaiting completion. +pub struct PipelineHandles { + download_handle: tokio::task::JoinHandle<()>, + clone_handle: tokio::task::JoinHandle<()>, +} + +impl PipelineHandles { + /// Wait for all pipeline workers to complete. + pub async fn await_completion(self) { + let _ = self.download_handle.await; + let _ = self.clone_handle.await; + } +} + +/// Start download and clone pipeline workers, returning handles to await completion. +pub fn start_workers(channels: PipelineChannels, cwd: PathBuf) -> PipelineHandles { + let download_handle = tokio::spawn(async move { + let mut rx = channels.download_rx; + while let Some(info) = rx.recv().await { + let Some(tarball_url) = info.tarball_url else { + continue; + }; + let name = info.name; + let version = info.version; + tokio::spawn(async move { + download_to_cache(&name, &version, &tarball_url).await; + }); + } + }); + + let clone_handle = tokio::spawn(async move { + let mut rx = channels.clone_rx; + while let Some(msg) = rx.recv().await { + let Some(tarball_url) = msg.info.tarball_url else { + continue; + }; + let name = msg.info.name; + let version = msg.info.version; + let target = cwd.join(&msg.path); + let parent_path = msg.parent_path.map(|p| cwd.join(&p)); + tokio::spawn(async move { + if let Some(ref parent) = parent_path { + wait_clone_if_pending(&parent.to_string_lossy()).await; + } + clone_package_once(&name, &version, &tarball_url, &target).await; + }); + } + }); + + PipelineHandles { + download_handle, + clone_handle, + } +} diff --git a/crates/pm/src/util/cloner.rs b/crates/pm/src/util/cloner.rs index cc44e7e5d9..6b3575addc 100644 --- a/crates/pm/src/util/cloner.rs +++ b/crates/pm/src/util/cloner.rs @@ -1,13 +1,75 @@ use anyhow::{Context, Result}; +use once_cell::sync::Lazy; use std::path::{Path, PathBuf}; +use std::sync::atomic::{AtomicUsize, Ordering}; +use super::downloader::download_to_cache; use super::json::load_package_json_from_path; -#[cfg(any(target_os = "macos", target_os = "linux"))] +use super::oncemap::OnceMap; use super::retry::create_retry_strategy; use crate::fs; -#[cfg(any(target_os = "macos", target_os = "linux"))] use tokio_retry::Retry; +/// Global clone cache shared between pipeline and install phases. +/// Key: target path, Value: () +static CLONE_CACHE: Lazy> = Lazy::new(OnceMap::new); + +/// Number of clones completed. +static CLONE_COUNT: AtomicUsize = AtomicUsize::new(0); + +/// Returns the number of fresh clones performed. +pub fn clone_count() -> usize { + CLONE_COUNT.load(Ordering::Relaxed) +} + +/// Wait for a pending clone at the given target path to complete (if any). +/// +/// Used by the pipeline clone worker to ensure parent packages are +/// cloned before their children. +pub async fn wait_clone_if_pending(target_path: &str) { + CLONE_CACHE.wait_if_pending(&target_path.to_string()).await; +} + +/// Clone a package to target path, downloading to cache first if needed. +/// +/// Uses global `OnceMap` for deduplication: the same target path is only cloned once, +/// even when called concurrently from pipeline workers and the install phase. +pub async fn clone_package_once( + name: &str, + version: &str, + tarball_url: &str, + target_path: &Path, +) -> bool { + let key = target_path.to_string_lossy().to_string(); + let name = name.to_string(); + let version = version.to_string(); + let tarball_url = tarball_url.to_string(); + let target_path = target_path.to_path_buf(); + + CLONE_CACHE + .get_or_init(key, || async move { + let cache_path = download_to_cache(&name, &version, &tarball_url).await?; + clone_package(&cache_path, &target_path, &name, &version) + .await + .inspect_err(|e| { + tracing::warn!( + "Clone failed: {}@{} to {}: {}", + name, + version, + target_path.display(), + e + ) + }) + .ok()?; + + CLONE_COUNT.fetch_add(1, Ordering::Relaxed); + tracing::debug!("Cloned: {}@{} to {}", name, version, target_path.display()); + Some(()) + }) + .await + .is_some() +} + #[cfg(target_os = "macos")] use libc::clonefile; #[cfg(target_os = "macos")] @@ -15,37 +77,42 @@ use std::ffi::CString; #[cfg(target_os = "macos")] use std::os::unix::ffi::OsStrExt; -#[cfg(target_os = "linux")] -mod linux_clone { +#[cfg(not(target_os = "macos"))] +mod hardlink_clone { use anyhow::{Context, Result}; + use std::collections::HashSet; use std::fs; use std::io; - use std::path::Path; + use std::path::{Path, PathBuf}; + + struct CloneEntry { + src: PathBuf, + dst: PathBuf, + } - // Check if the package has install scripts (sync version) fn has_install_script_sync(src: &Path) -> bool { src.parent().is_some_and(|parent| { fs::metadata(parent.join("_hasInstallScript")).is_ok_and(|m| m.is_file()) }) } - // Copy a single file, preserving permissions (sync version) fn copy_file_sync(src: &Path, dst: &Path) -> io::Result<()> { fs::copy(src, dst)?; - // Preserve permissions - let src_perms = fs::metadata(src)?.permissions(); - fs::set_permissions(dst, src_perms)?; + #[cfg(unix)] + { + let src_perms = fs::metadata(src)?.permissions(); + fs::set_permissions(dst, src_perms)?; + } Ok(()) } - // Recursive directory copy with hardlink strategy (sync version) - fn clone_dir_sync(src: &Path, dst: &Path, use_copy: bool) -> io::Result<()> { - // Create destination directory - if let Err(e) = fs::create_dir(dst) - && e.kind() != io::ErrorKind::AlreadyExists - { - return Err(e); - } + fn collect_entries( + src: &Path, + dst: &Path, + files: &mut Vec, + dirs: &mut Vec, + ) -> io::Result<()> { + dirs.push(dst.to_path_buf()); for entry in fs::read_dir(src)? { let entry = entry?; @@ -54,25 +121,21 @@ mod linux_clone { let target_path = dst.join(&file_name); if entry.file_type()?.is_dir() { - clone_dir_sync(&entry_path, &target_path, use_copy)?; - } else if use_copy { - // For packages with install scripts, always copy - copy_file_sync(&entry_path, &target_path)?; + collect_entries(&entry_path, &target_path, files, dirs)?; } else { - // Try hardlink first - if fs::hard_link(&entry_path, &target_path).is_err() { - // Fallback to copy - copy_file_sync(&entry_path, &target_path)?; - } + files.push(CloneEntry { + src: entry_path, + dst: target_path, + }); } } Ok(()) } - // Entry point: clone directory using spawn_blocking + /// Clone directory using spawn_blocking for sync I/O. + /// Uses hardlink when possible, falls back to copy. pub async fn clone_dir(src: &Path, dst: &Path) -> Result<()> { - // Pre-format for error message (avoid clone) let err_msg = format!("Failed to clone {} to {}", src.display(), dst.display()); let src = src.to_path_buf(); let dst = dst.to_path_buf(); @@ -85,143 +148,35 @@ mod linux_clone { )); } - // Create destination directory (entry point) - fs::create_dir_all(&dst)?; - - // Check if the package has install scripts let use_copy = has_install_script_sync(&src); - // Clone directory recursively - clone_dir_sync(&src, &dst, use_copy) - }) - .await? - .with_context(|| err_msg) - } -} - -#[cfg(target_os = "windows")] -mod windows_clone { - use crate::fs; - use anyhow::{Context, Result}; - use std::path::Path; - - // Copy a single file using regular copy - async fn fast_copy_file(src: &Path, dst: &Path) -> Result<()> { - crate::fs::copy(src, dst).await.with_context(|| { - format!( - "Failed to copy file from {} to {}", - src.display(), - dst.display() - ) - })?; - Ok(()) - } - - // Fast copy directory (internal recursive version) - async fn fast_copy_inner(src: &Path, dst: &Path) -> Result<()> { - // Use create_dir since parent already exists - if let Err(e) = tokio::fs::create_dir(dst).await - && e.kind() != std::io::ErrorKind::AlreadyExists - { - return Err(e.into()); - } - - // Copy all files in the directory - let mut read_dir = fs::read_dir(src).await?; - while let Some(entry) = read_dir.next_entry().await? { - let entry_path = entry.path(); - let file_name = entry_path.file_name().unwrap(); - let target_path = dst.join(file_name); - - if entry.metadata().await?.is_dir() { - Box::pin(fast_copy_inner(&entry_path, &target_path)).await?; - } else { - fast_copy_file(&entry_path, &target_path).await?; - } - } - Ok(()) - } - - // Fast copy entry point - async fn fast_copy(src: &Path, dst: &Path) -> Result<()> { - fs::create_dir_all(dst).await?; - fast_copy_inner(src, dst).await - } - - // Check if the package has install scripts - async fn has_install_script(src: &Path) -> bool { - if let Some(parent) = src.parent() { - let flag_path = parent.join("_hasInstallScript"); - if let Ok(metadata) = fs::metadata(&flag_path).await { - return metadata.is_file(); + // Phase 1: Collect all files and directories + let mut files = Vec::new(); + let mut dirs = Vec::new(); + collect_entries(&src, &dst, &mut files, &mut dirs)?; + + // Phase 2: Create all directories + let mut created_dirs = HashSet::new(); + for dir in &dirs { + if created_dirs.insert(dir.clone()) + && let Err(e) = fs::create_dir_all(dir) + && e.kind() != io::ErrorKind::AlreadyExists + { + return Err(e); + } } - } - false - } - - // Internal recursive clone (parent directory already exists) - async fn clone_dir_inner(src: &Path, dst: &Path) -> Result<()> { - // Use create_dir since parent already exists - if let Err(e) = tokio::fs::create_dir(dst).await - && e.kind() != std::io::ErrorKind::AlreadyExists - { - return Err(e.into()); - } - - let mut read_dir = fs::read_dir(src).await?; - while let Some(entry) = read_dir.next_entry().await? { - let entry_path = entry.path(); - let file_name = entry_path.file_name().unwrap(); - let target_path = dst.join(file_name); - if entry.metadata().await?.is_dir() { - Box::pin(clone_dir_inner(&entry_path, &target_path)).await?; - } else { - // Try hardlink first for files in packages without install scripts - if let Err(e) = fs::hard_link(&entry_path, &target_path).await { - tracing::debug!( - "Failed to create hardlink for file from {} to {}: {}, using regular copy", - entry_path.display(), - target_path.display(), - e - ); - // If hardlink fails, fallback to regular copy - crate::fs::copy(&entry_path, &target_path) - .await - .with_context(|| { - format!( - "Failed to copy file from {} to {}", - entry_path.display(), - target_path.display() - ) - })?; + // Phase 3: Clone files (hardlink first, fallback to copy) + for entry in &files { + if !use_copy && fs::hard_link(&entry.src, &entry.dst).is_ok() { + continue; } + copy_file_sync(&entry.src, &entry.dst)?; } - } - - Ok(()) - } - - pub async fn clone_dir(src: &Path, dst: &Path) -> Result<()> { - if !fs::metadata(src).await?.is_dir() { - return Err(anyhow::anyhow!("Source is not a directory")); - } - - // Create destination directory (entry point, use create_dir_all) - fs::create_dir_all(dst).await?; - - // Check if the package has install scripts - if has_install_script(src).await { - tracing::debug!( - "Package has install scripts, using fast copy for directory {} to {}", - src.display(), - dst.display() - ); - return fast_copy_inner(src, dst).await; - } - - // For directories without install scripts, use hardlink strategy - clone_dir_inner(src, dst).await + Ok(()) + }) + .await? + .with_context(|| err_msg) } } @@ -412,36 +367,16 @@ async fn clone(src: &Path, dst: &Path, find_real: bool) -> Result<()> { .await?; } - #[cfg(target_os = "linux")] - { - Retry::spawn(create_retry_strategy(), || async { - linux_clone::clone_dir(&real_src, dst).await?; - tracing::debug!("clone {} to {} success", real_src.display(), dst.display()); - Ok::<(), anyhow::Error>(()) - }) - .await?; - } - - #[cfg(target_os = "windows")] + #[cfg(not(target_os = "macos"))] { - use super::retry::create_retry_strategy; - use tokio_retry::Retry; - Retry::spawn(create_retry_strategy(), || async { - windows_clone::clone_dir(&real_src, dst).await?; + hardlink_clone::clone_dir(&real_src, dst).await?; tracing::debug!("clone {} to {} success", real_src.display(), dst.display()); Ok::<(), anyhow::Error>(()) }) .await?; } - #[cfg(not(any(target_os = "macos", target_os = "linux", target_os = "windows")))] - { - // For other platforms, use standard recursive copy - copy_dir_all(&real_src, dst).await?; - tracing::debug!("clone {} to {} success", real_src.display(), dst.display()); - } - Ok(()) } @@ -480,27 +415,6 @@ pub async fn clone_package(src: &Path, dst: &Path, name: &str, version: &str) -> } } -// Standard directory copy for non-macOS/Linux/Windows platforms -#[cfg(not(any(target_os = "macos", target_os = "linux", target_os = "windows")))] -async fn copy_dir_all(src: &Path, dst: &Path) -> Result<()> { - fs::create_dir_all(dst).await?; - let mut entries = fs::read_dir(src).await?; - - while let Some(entry) = entries.next_entry().await? { - let ty = entry.file_type().await?; - let src_path = entry.path(); - let dst_path = dst.join(entry.file_name()); - - if ty.is_dir() { - Box::pin(copy_dir_all(&src_path, &dst_path)).await?; - } else { - fs::copy(&src_path, &dst_path).await?; - } - } - - Ok(()) -} - #[cfg(test)] mod tests { use super::*; @@ -845,8 +759,8 @@ mod tests { Ok(()) } - #[cfg(target_os = "linux")] - mod linux_tests { + #[cfg(not(target_os = "macos"))] + mod hardlink_clone_tests { use super::*; use crate::fs::File; use tokio::io::AsyncReadExt; @@ -870,7 +784,7 @@ mod tests { .await?; // Perform clone operation - linux_clone::clone_dir(&src_dir, &dst_dir).await?; + hardlink_clone::clone_dir(&src_dir, &dst_dir).await?; // Verify clone result assert!(validate_directory(&src_dir, &dst_dir).await?); @@ -912,7 +826,7 @@ mod tests { .await?; // Perform clone operation - linux_clone::clone_dir(&src_dir, &dst_dir).await?; + hardlink_clone::clone_dir(&src_dir, &dst_dir).await?; // Verify clone result assert!(validate_directory(&src_dir, &dst_dir).await?); @@ -935,7 +849,7 @@ mod tests { let dst_dir = temp.path().join("dst"); // Test case when source directory doesn't exist - let result = linux_clone::clone_dir(&src_dir, &dst_dir).await; + let result = hardlink_clone::clone_dir(&src_dir, &dst_dir).await; assert!(result.is_err()); assert_eq!( result @@ -949,12 +863,13 @@ mod tests { // Test case when source path is a file instead of a directory create_test_file(temp.path(), "not_a_dir", b"content").await?; let result = - linux_clone::clone_dir(temp.path().join("not_a_dir").as_ref(), &dst_dir).await; + hardlink_clone::clone_dir(temp.path().join("not_a_dir").as_ref(), &dst_dir).await; assert!(result.is_err()); Ok(()) } + #[cfg(unix)] #[tokio::test] async fn test_fast_copy_preserves_permissions() -> Result<()> { use std::os::unix::fs::PermissionsExt; @@ -978,7 +893,7 @@ mod tests { assert_eq!(src_metadata.permissions().mode() & 0o777, 0o755); // Use clone_dir to trigger fast_copy_file path (no _hasInstallScript flag) - linux_clone::clone_dir(&src_dir, &dst_dir).await?; + hardlink_clone::clone_dir(&src_dir, &dst_dir).await?; // Verify destination file has same permissions let dst_file = dst_dir.join("executable.sh"); @@ -992,6 +907,7 @@ mod tests { Ok(()) } + #[cfg(unix)] #[tokio::test] async fn test_fast_copy_preserves_read_only_permissions() -> Result<()> { use std::os::unix::fs::PermissionsExt; @@ -1015,7 +931,7 @@ mod tests { assert_eq!(src_metadata.permissions().mode() & 0o777, 0o444); // Use clone_dir to trigger fast_copy_file path - linux_clone::clone_dir(&src_dir, &dst_dir).await?; + hardlink_clone::clone_dir(&src_dir, &dst_dir).await?; // Verify destination file has same permissions let dst_file = dst_dir.join("readonly.txt"); @@ -1029,6 +945,7 @@ mod tests { Ok(()) } + #[cfg(unix)] #[tokio::test] async fn test_clone_dir_preserves_executable_in_subdirs() -> Result<()> { use std::os::unix::fs::PermissionsExt; @@ -1062,7 +979,7 @@ mod tests { fs::set_permissions(&data_path, perms).await?; // Clone directory - linux_clone::clone_dir(&src_dir, &dst_dir).await?; + hardlink_clone::clone_dir(&src_dir, &dst_dir).await?; // Verify script.sh has executable permissions let dst_script = dst_dir.join("bin/script.sh"); diff --git a/crates/pm/src/util/config.rs b/crates/pm/src/util/config.rs index 1956dc948a..dc99b56e4a 100644 --- a/crates/pm/src/util/config.rs +++ b/crates/pm/src/util/config.rs @@ -254,6 +254,10 @@ pub async fn get_manifests_concurrency_limit() -> usize { MANIFESTS_CONCURRENCY_LIMIT.get().await } +pub fn get_manifests_concurrency_limit_sync() -> usize { + MANIFESTS_CONCURRENCY_LIMIT.get_sync() +} + pub async fn set_cache_dir(cache_dir: Option) { // Priority: CLI argument > UTOO_CACHE_DIR env > config > default let final_cache_dir = if let Some(dir) = cache_dir { diff --git a/crates/pm/src/util/downloader.rs b/crates/pm/src/util/downloader.rs index 8202833c24..e96216db50 100644 --- a/crates/pm/src/util/downloader.rs +++ b/crates/pm/src/util/downloader.rs @@ -1,79 +1,129 @@ use anyhow::{Context, Result}; -use async_compression::tokio::bufread::GzipDecoder; -use futures::StreamExt; +use bytes::Bytes; use once_cell::sync::Lazy; +use reqwest::Client; use reqwest::StatusCode; -use reqwest::{Client, Response}; -#[cfg(unix)] -use std::os::unix::fs::PermissionsExt; -use std::path::Path; -use tokio::{fs::File, io::AsyncReadExt}; +use std::path::PathBuf; +use std::sync::OnceLock; +use std::sync::atomic::{AtomicU32, AtomicUsize, Ordering}; +use tokio::sync::Semaphore; use tokio_retry::RetryIf; -use tokio_tar::Archive; -use tokio_util::io::StreamReader; -use super::retry::build_dns_cached_client; -use super::retry::{RetryableError, create_retry_strategy}; +use super::cache::get_cache_dir; +use super::config::get_manifests_concurrency_limit_sync; +use super::extractor::extract_and_write; +use super::oncemap::OnceMap; +use super::retry::{RetryableError, build_dns_cached_client, create_retry_strategy}; -use dashmap::DashMap; -use std::collections::hash_map::DefaultHasher; -use std::hash::{Hash, Hasher}; -use std::sync::Arc; -use tokio::sync::Mutex; - -// Global downloader client - no pool limit, concurrency controlled by semaphore +// Global downloader client - no pool limit, concurrency controlled by OnceMap static DOWNLOADER_CLIENT: Lazy = Lazy::new(build_dns_cached_client); -static DOWNLOAD_LOCKS: Lazy>>> = Lazy::new(DashMap::new); +/// Global download cache shared between pipeline and install phases. +/// Key: "name@version", Value: cache path. +static DOWNLOAD_CACHE: Lazy> = Lazy::new(OnceMap::new); + +/// Semaphore controlling concurrent download count. +static DOWNLOAD_SEMAPHORE: OnceLock = OnceLock::new(); + +/// Number of fresh downloads (not cache hits). +static DOWNLOAD_COUNT: AtomicUsize = AtomicUsize::new(0); -fn lock_key(url: &str, dest: &Path) -> u64 { - let mut hasher = DefaultHasher::new(); - url.hash(&mut hasher); - dest.to_string_lossy().hash(&mut hasher); - hasher.finish() +/// Returns the number of fresh downloads performed. +pub fn download_count() -> usize { + DOWNLOAD_COUNT.load(Ordering::Relaxed) } -pub async fn download(url: &str, dest: &Path) -> Result<()> { - let start = std::time::Instant::now(); - let key = lock_key(url, dest); - let lock = DOWNLOAD_LOCKS - .entry(key) - .or_insert_with(|| Arc::new(Mutex::new(()))) - .clone(); - let _guard = lock.lock().await; +/// Download a package tarball to the global cache directory, returning the cache path. +/// +/// Uses `OnceMap` to deduplicate: the same `name@version` is only downloaded once, +/// even when called concurrently from multiple tasks (pipeline workers, install phase, etc.). +pub async fn download_to_cache(name: &str, version: &str, tarball_url: &str) -> Option { + let key = format!("{}@{}", name, version); + let cache_dir = get_cache_dir(); + let name = name.to_string(); + let version = version.to_string(); + let tarball_url = tarball_url.to_string(); + + DOWNLOAD_CACHE + .get_or_init(key, || async move { + let cache_path = cache_dir.join(&name).join(&version); + + // Fast path: already extracted in cache + if crate::fs::try_exists(&cache_path.join("_resolved")) + .await + .unwrap_or(false) + { + tracing::debug!("Cache hit: {}@{}", name, version); + return Some(cache_path); + } - let resolved_path = dest.join("_resolved"); - if crate::fs::try_exists(&resolved_path).await? { - tracing::debug!("Download skipped, already resolved: {}", dest.display()); - return Ok(()); - } + // Download (semaphore controlled) + let semaphore = DOWNLOAD_SEMAPHORE + .get_or_init(|| Semaphore::new(get_manifests_concurrency_limit_sync())); + let _permit = semaphore.acquire().await.ok()?; + let bytes = download_bytes(&tarball_url) + .await + .inspect_err(|e| tracing::warn!("Download failed: {}@{}: {}", name, version, e)) + .ok()?; + + // Extract + extract_and_write(bytes, &cache_path) + .await + .inspect_err(|e| tracing::warn!("Extract failed: {}@{}: {}", name, version, e)) + .ok()?; + DOWNLOAD_COUNT.fetch_add(1, Ordering::Relaxed); + tracing::debug!("Downloaded: {}@{}", name, version); + Some(cache_path) + }) + .await + .as_deref() + .cloned() +} + +/// Download tarball bytes with retries (network phase only). +pub async fn download_bytes(url: &str) -> Result { + let retry_count = AtomicU32::new(0); RetryIf::spawn( create_retry_strategy(), || async { - let response = DOWNLOADER_CLIENT - .get(url) - .send() - .await - .with_context(|| format!("Failed to send HTTP request to {url}")) - .map_err(|e| RetryableError::Temporary(format!("Network error: {e}")))?; + let attempt = retry_count.fetch_add(1, Ordering::Relaxed); + + let response = match DOWNLOADER_CLIENT.get(url).send().await { + Ok(r) => r, + Err(e) => { + tracing::warn!( + "Retry {}/10 - Network error: {}, url: {}", + attempt + 1, + e, + url + ); + return Err(RetryableError::Temporary(format!("Network error: {e}"))); + } + }; match response.status() { StatusCode::OK => { - if let Err(e) = try_unpack_stream_direct(response, dest).await { - tracing::debug!("Stream unpacking failed {}: {:#}", dest.display(), e); - return Err(RetryableError::Temporary(format!( - "Network error during streaming: {e:#}" - ))); + let bytes = response.bytes().await.map_err(|e| { + tracing::warn!( + "Retry {}/10 - Stream error: {}, url: {}", + attempt + 1, + e, + url + ); + RetryableError::Temporary(format!("Stream error: {e}")) + })?; + if attempt > 0 { + tracing::info!("Retry succeeded on attempt {}, url: {}", attempt + 1, url); } - Ok(()) + Ok(bytes) } StatusCode::NOT_FOUND => { tracing::debug!("URL not found {url}"); Err(RetryableError::Permanent(format!("URL not found {url}"))) } status => { - tracing::debug!("Error: {status}, url: {url}, retrying"); + tracing::warn!("Retry {}/10 - HTTP {}, url: {}", attempt + 1, status, url); Err(RetryableError::Temporary(format!( "HTTP error: {status}, url: {url}" ))) @@ -83,222 +133,7 @@ pub async fn download(url: &str, dest: &Path) -> Result<()> { |e: &RetryableError| matches!(e, RetryableError::Temporary(_)), ) .await - .context("Download failed after retries")?; - - let duration = start.elapsed(); - tracing::debug!("Download task took: {duration:?}, url: {url:?}"); - Ok(()) -} - -// Stream-based unpacking directly from HTTP Response -async fn try_unpack_stream_direct(response: Response, dest: &Path) -> Result<()> { - use std::sync::Arc; - use tokio::sync::{Semaphore, mpsc}; - - crate::fs::create_dir_all(dest) - .await - .with_context(|| format!("Failed to create destination directory: {}", dest.display()))?; - - // Convert HTTP response stream to AsyncRead for streaming processing - let stream = response - .bytes_stream() - .map(|result| result.map_err(std::io::Error::other)); - let stream_reader = StreamReader::new(stream); - - // Create pipeline processing channels - let (entry_tx, mut entry_rx) = mpsc::channel::(500); - - let dest = dest.to_path_buf(); - - // Stage 1: Streaming tar extraction - let extraction_task = { - let entry_tx = entry_tx.clone(); - let dest = dest.clone(); - - tokio::spawn(async move { - // Create streaming gzip decoder - let gzip_decoder = GzipDecoder::new(stream_reader); - let mut tar_archive = Archive::new(gzip_decoder); - let mut entries = tar_archive.entries()?; - - while let Some(entry_result) = entries.next().await { - 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(); - let full_path = dest.join(&path); - let is_dir = entry.header().entry_type().is_dir(); - - // Only process files, skip directories (they'll be created when writing files) - if !is_dir { - // Stream file content - let mut content = Vec::new(); - entry - .read_to_end(&mut content) - .await - .with_context(|| format!("Failed to read tar entry: {}", path.display()))?; - - // Extract file permission mode - let mode = entry.header().mode().unwrap_or(0o644); - - let size = content.len(); - let extracted_entry = ExtractedEntry { - path: full_path, - content, - size, - mode, - }; - - if entry_tx.send(extracted_entry).await.is_err() { - break; - } - } - } - - Ok::<(), anyhow::Error>(()) - }) - }; - - // Stage 2: Concurrent file writing with cached directory creation - let file_writing_task = { - tokio::spawn(async move { - use dashmap::DashSet; - - let semaphore = Arc::new(Semaphore::new(16)); - let created_dirs = Arc::new(DashSet::::new()); - let mut write_tasks = Vec::new(); - let mut batch_size = 0; - let mut total_bytes = 0; - const MAX_BATCH_SIZE: usize = 100; - const MAX_BATCH_BYTES: usize = 50 * 1024 * 1024; // 50MB - - while let Some(entry) = entry_rx.recv().await { - let semaphore = Arc::clone(&semaphore); - let created_dirs = Arc::clone(&created_dirs); - batch_size += 1; - total_bytes += entry.size; - - let task = tokio::spawn(async move { - let _permit = semaphore.acquire().await.unwrap(); - - // Ensure parent directory exists using cache - if let Some(parent) = entry.path.parent() { - let parent_path = parent.to_path_buf(); - - // Check cache first to avoid duplicate directory creation - if !created_dirs.contains(&parent_path) { - if let Err(e) = crate::fs::create_dir_all(&parent_path).await { - tracing::debug!( - "Failed to create parent dir {}: {}", - parent_path.display(), - e - ); - return Err(anyhow::anyhow!( - "Failed to create parent directory: {e}" - ) - .context(format!("Parent directory: {}", parent_path.display()))); - } - - created_dirs.insert(parent_path); - } - } - - // Write file content - if let Err(e) = crate::fs::write(&entry.path, &entry.content).await { - tracing::debug!("Failed to write file {}: {}", entry.path.display(), e); - return Err(anyhow::anyhow!("Write failed: {e}") - .context(format!("File path: {}", entry.path.display()))); - } - - // Set original file permissions from tar entry (Unix only) - set_file_permissions(&entry.path, entry.mode).await?; - - Ok::<(), anyhow::Error>(()) - }); - - write_tasks.push(task); - - // Process in batches to manage memory and concurrency - if batch_size >= MAX_BATCH_SIZE - || total_bytes >= MAX_BATCH_BYTES - || entry_rx.is_empty() - { - for task in write_tasks.drain(..) { - task.await??; - } - batch_size = 0; - total_bytes = 0; - } - } - - // Wait for remaining tasks - for task in write_tasks { - task.await??; - } - - Ok::<(), anyhow::Error>(()) - }) - }; - - // Close sender channel - drop(entry_tx); - - // Wait for both stages to complete - let (extract_result, write_result) = tokio::try_join!(extraction_task, file_writing_task)?; - - extract_result?; - write_result?; - - // Set directory permissions and create resolution marker - set_dir_permissions(&dest).await?; - File::create(&dest.join("_resolved")) - .await - .with_context(|| format!("Failed to create resolution marker in: {}", dest.display()))?; - - Ok(()) -} - -/// Set file permissions (cross-platform) -#[cfg(unix)] -async fn set_file_permissions(path: &Path, mode: u32) -> Result<()> { - use std::fs::Permissions; - let permissions = Permissions::from_mode(mode); - crate::fs::set_permissions(path, permissions) - .await - .with_context(|| format!("Failed to set permissions for: {}", path.display()))?; - Ok(()) -} - -#[cfg(not(unix))] -async fn set_file_permissions(_path: &Path, _mode: u32) -> Result<()> { - // Windows doesn't need Unix-style permissions - Ok(()) -} - -/// Set directory permissions (cross-platform) -#[cfg(unix)] -async fn set_dir_permissions(path: &Path) -> Result<()> { - use std::fs::Permissions; - let permissions = Permissions::from_mode(0o755); - crate::fs::set_permissions(path, permissions) - .await - .with_context(|| format!("Failed to set directory permissions: {}", path.display()))?; - Ok(()) -} - -#[cfg(not(unix))] -async fn set_dir_permissions(_path: &Path) -> Result<()> { - // Windows doesn't need Unix-style permissions - Ok(()) -} - -#[derive(Debug)] -struct ExtractedEntry { - path: std::path::PathBuf, - content: Vec, - size: usize, - mode: u32, // File permission mode + .context("Download failed after retries") } #[cfg(test)] @@ -306,11 +141,9 @@ mod tests { use super::*; use flate2::Compression; use flate2::write::GzEncoder; - use mockito::Server; use std::io::Write; use tar::Builder; use tempfile::TempDir; - use tokio::task; // Helper to create a simple tar.gz archive in memory fn create_tar_gz() -> Vec { @@ -331,42 +164,38 @@ mod tests { } #[tokio::test] - async fn test_download_idempotent() { + async fn test_extract_and_write() { let tar_gz = create_tar_gz(); - let mut server = Server::new_async().await; - let _m = server - .mock("GET", "/pkg.tgz") - .with_status(200) - .with_header("content-type", "application/gzip") - .with_body(tar_gz.clone()) - .expect(1) - .create_async() - .await; - - let url = format!("{}/pkg.tgz", server.url()); let temp_dir = TempDir::new().unwrap(); let dest = temp_dir.path().join("pkg"); - let n = 8; - let mut handles = Vec::new(); - for _ in 0..n { - let url = url.clone(); - let dest = dest.clone(); - handles.push(task::spawn(async move { - download(&url, &dest).await.unwrap(); - })); - } - for h in handles { - h.await.unwrap(); - } + + extract_and_write(Bytes::from(tar_gz), &dest).await.unwrap(); + // _resolved file should exist assert!(dest.join("_resolved").exists()); // Extracted file should exist assert!(dest.join("file.txt").exists()); - // All concurrent calls should succeed and not corrupt the output let content = crate::fs::read_to_string(dest.join("file.txt")) .await .unwrap(); assert_eq!(content, "hello world"); - _m.assert(); + } + + #[tokio::test] + async fn test_extract_and_write_idempotent() { + let tar_gz = create_tar_gz(); + let temp_dir = TempDir::new().unwrap(); + let dest = temp_dir.path().join("pkg"); + + // First extraction + extract_and_write(Bytes::from(tar_gz.clone()), &dest) + .await + .unwrap(); + + // Second extraction should skip (already resolved) + extract_and_write(Bytes::from(tar_gz), &dest).await.unwrap(); + + assert!(dest.join("_resolved").exists()); + assert!(dest.join("file.txt").exists()); } } diff --git a/crates/pm/src/util/extractor.rs b/crates/pm/src/util/extractor.rs new file mode 100644 index 0000000000..06adf4a2a9 --- /dev/null +++ b/crates/pm/src/util/extractor.rs @@ -0,0 +1,158 @@ +use anyhow::{Context, Result}; +use bytes::Bytes; +#[cfg(unix)] +use std::os::unix::fs::PermissionsExt; +use std::path::{Path, PathBuf}; + +const MIN_ESTIMATED_SIZE: usize = 16; +const MAX_ESTIMATED_SIZE: usize = 512 * 1024 * 1024; // 512MB +const DECOMPRESSION_RETRY_FACTOR: usize = 4; + +/// Extract gzip tarball and write to destination. +/// +/// Skips if `_resolved` marker already exists (warm cache). +pub async fn extract_and_write(gzip_bytes: Bytes, dest: &Path) -> Result<()> { + // Check if already resolved (warm cache scenario) + let resolved_path = dest.join("_resolved"); + if crate::fs::try_exists(&resolved_path).await? { + tracing::debug!("Extract skipped, already resolved: {}", dest.display()); + return Ok(()); + } + + crate::fs::create_dir_all(dest) + .await + .with_context(|| format!("Failed to create destination directory: {}", dest.display()))?; + + extract_tarball(gzip_bytes, dest).await +} + +/// Estimate uncompressed size from gzip footer (last 4 bytes store original size mod 2^32). +fn estimate_uncompressed_size(gzip_data: &[u8]) -> usize { + if gzip_data.len() < 4 { + return gzip_data.len() * 10; // fallback estimate + } + let last_4 = &gzip_data[gzip_data.len() - 4..]; + let size = u32::from_le_bytes([last_4[0], last_4[1], last_4[2], last_4[3]]) as usize; + // Sanity check: if size is 0 or too small, use a reasonable estimate + if !(MIN_ESTIMATED_SIZE..=MAX_ESTIMATED_SIZE).contains(&size) { + gzip_data.len() * 10 + } else { + size + } +} + +struct ExtractedEntry { + path: PathBuf, + content: Vec, + mode: u32, +} + +/// Extract tarball using libdeflate for decompression + rayon for parallel writes. +/// +/// Uses rayon::spawn (not tokio blocking pool) to avoid thread storms. +/// Rayon's global pool is configured with sufficient stack size at startup. +async fn extract_tarball(gzip_bytes: Bytes, dest: &Path) -> Result<()> { + let estimated_size = estimate_uncompressed_size(&gzip_bytes); + let dest_owned = dest.to_path_buf(); + + let (tx, rx) = tokio::sync::oneshot::channel(); + + rayon::spawn(move || { + let result = extract_tarball_sync(gzip_bytes, estimated_size, &dest_owned); + let _ = tx.send(result); + }); + + rx.await.with_context(|| "Extract task panicked")? +} + +/// Synchronous extraction: decompress + parse + parallel write, all on rayon. +fn extract_tarball_sync(gzip_bytes: Bytes, estimated_size: usize, dest: &Path) -> Result<()> { + use rayon::prelude::*; + use std::collections::HashSet; + use std::fs; + use std::io::{Cursor, Read, Write}; + + // Decompress gzip using libdeflate + let mut output = vec![0u8; estimated_size]; + + let mut decompressor = libdeflater::Decompressor::new(); + + let actual_size = match decompressor.gzip_decompress(&gzip_bytes, &mut output) { + Ok(size) => size, + Err(libdeflater::DecompressionError::InsufficientSpace) => { + let new_size = estimated_size * DECOMPRESSION_RETRY_FACTOR; + output.resize(new_size, 0); + decompressor + .gzip_decompress(&gzip_bytes, &mut output) + .with_context(|| "gzip decompression failed")? + } + Err(e) => return Err(anyhow::anyhow!("gzip decompression failed: {}", e)), + }; + 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(); + 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()))?; + + let mode = entry.header().mode().unwrap_or(0o644); + entries.push(ExtractedEntry { + path: full_path, + content, + mode, + }); + } + } + + // Create directories + let mut created_dirs = HashSet::new(); + for entry in entries.iter() { + if let Some(parent) = entry.path.parent() + && created_dirs.insert(parent.to_path_buf()) + { + fs::create_dir_all(parent).ok(); + } + } + + // Write files in parallel using rayon + 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) + .with_context(|| format!("Failed to write: {}", entry.path.display()))?; + + #[cfg(unix)] + { + let perms = fs::Permissions::from_mode(entry.mode); + fs::set_permissions(&entry.path, perms).ok(); + } + Ok(()) + })?; + + // Set directory permissions + #[cfg(unix)] + { + let perms = fs::Permissions::from_mode(0o755); + fs::set_permissions(dest, perms).ok(); + } + + // Create resolution marker + fs::File::create(dest.join("_resolved")) + .with_context(|| format!("Failed to create resolution marker in: {}", dest.display()))?; + + Ok(()) +} diff --git a/crates/pm/src/util/mod.rs b/crates/pm/src/util/mod.rs index 03776bfe02..d512e4b4ff 100644 --- a/crates/pm/src/util/mod.rs +++ b/crates/pm/src/util/mod.rs @@ -3,6 +3,7 @@ pub mod cache; pub mod cloner; pub mod config; pub mod downloader; +pub mod extractor; pub mod format_print; pub mod json; pub mod linker; diff --git a/crates/pm/src/util/oncemap.rs b/crates/pm/src/util/oncemap.rs index 6e2f81def7..c189f61cc9 100644 --- a/crates/pm/src/util/oncemap.rs +++ b/crates/pm/src/util/oncemap.rs @@ -66,6 +66,9 @@ enum Value { /// ``` pub struct OnceMap { map: DashMap>, + /// Waiters for keys that don't exist yet. + /// When get_or_init/complete finishes a key, it notifies and removes the entry. + waiters: DashMap>, } impl Default for OnceMap @@ -85,6 +88,103 @@ where pub fn new() -> Self { Self { map: DashMap::new(), + waiters: DashMap::new(), + } + } + + /// Register a key as pending (Waiting state) without starting work. + /// Returns the Notify handle if newly registered, None if already exists. + pub fn register(&self, key: K) -> Option> { + use dashmap::mapref::entry::Entry; + match self.map.entry(key) { + Entry::Occupied(_) => None, + Entry::Vacant(vacant) => { + let notify = Arc::new(Notify::new()); + vacant.insert(Value::Waiting(Arc::clone(¬ify))); + Some(notify) + } + } + } + + /// Wait for a key to complete. + /// + /// - If the key exists and is Done, returns immediately. + /// - If the key exists and is Waiting, waits for it to complete. + /// - If the key doesn't exist yet, waits for it to be created and completed. + /// + /// This enables parent-ordering in the clone pipeline: a child package + /// can wait for its parent's clone to finish even if the parent's + /// `get_or_init` hasn't been called yet. + pub async fn wait_if_pending(&self, key: &K) { + // Fast path: key already exists + if let Some(entry) = self.map.get(key) { + match entry.value() { + Value::Done(_) => return, + Value::Waiting(notify) => { + let notify = Arc::clone(notify); + let notified = notify.notified(); + drop(entry); + + if let Some(entry) = self.map.get(key) + && matches!(entry.value(), Value::Done(_)) + { + return; + } + + notified.await; + return; + } + } + } + + // Key doesn't exist yet — register a creation waiter. + // get_or_init/complete will notify this when the key reaches Done (or fails). + let creation_notify = self + .waiters + .entry(key.clone()) + .or_insert_with(|| Arc::new(Notify::new())) + .clone(); + + // Register notified BEFORE double-checking to prevent missed notifications. + let notified = creation_notify.notified(); + + // Double-check: key might have been created and completed between + // the initial map.get and registering here. + if let Some(entry) = self.map.get(key) + && matches!(entry.value(), Value::Done(_)) + { + return; + } + + // Wait for the key to be created and completed + notified.await; + } + + /// Complete a pre-registered key with a value. + /// + /// This is used in conjunction with `register()` for the pre-registration pattern: + /// 1. Call `register(key)` synchronously to claim the key + /// 2. Do async work + /// 3. Call `complete(key, value, notify)` to store result and notify waiters + /// + /// The notify handle must be the one returned from `register()`. + pub fn complete(&self, key: K, value: Option, notify: Arc) { + match value { + Some(v) => { + self.map.insert(key.clone(), Value::Done(Arc::new(v))); + } + None => { + self.map.remove(&key); + } + } + self.notify_all(&key, ¬ify); + } + + /// Notify both the work-notify and any creation waiters for a key. + fn notify_all(&self, key: &K, notify: &Notify) { + notify.notify_waiters(); + if let Some((_, cn)) = self.waiters.remove(key) { + cn.notify_waiters(); } } @@ -161,20 +261,16 @@ where let result = init().await; // Update the map with the result - match result { - Some(value) => { - let arc_value = Arc::new(value); - self.map.insert(key, Value::Done(Arc::clone(&arc_value))); - notify.notify_waiters(); - Some(arc_value) - } - None => { - // Work failed, remove the entry so others can retry - self.map.remove(&key); - notify.notify_waiters(); - None - } + let arc_value = result.map(|v| { + let arc = Arc::new(v); + self.map.insert(key.clone(), Value::Done(Arc::clone(&arc))); + arc + }); + if arc_value.is_none() { + self.map.remove(&key); } + self.notify_all(&key, ¬ify); + arc_value } } diff --git a/crates/ruborist/src/lib.rs b/crates/ruborist/src/lib.rs index 95b7584f73..5506166896 100644 --- a/crates/ruborist/src/lib.rs +++ b/crates/ruborist/src/lib.rs @@ -66,7 +66,9 @@ pub mod builder { /// Progress events for build process. pub mod progress { - pub use crate::traits::progress::{BuildEvent, EventReceiver, NoopReceiver}; + pub use crate::traits::progress::{ + BuildEvent, EventReceiver, NoopReceiver, PackageTarballInfo, + }; } /// Platform compatibility checks. diff --git a/crates/ruborist/src/model/mod.rs b/crates/ruborist/src/model/mod.rs index c19c491fd1..a9df04cf94 100644 --- a/crates/ruborist/src/model/mod.rs +++ b/crates/ruborist/src/model/mod.rs @@ -7,6 +7,7 @@ pub mod node; pub mod override_rule; pub mod package_json; pub mod package_lock; +pub mod tarball_info; pub(crate) mod util; pub use util::parse_package_spec; diff --git a/crates/ruborist/src/model/tarball_info.rs b/crates/ruborist/src/model/tarball_info.rs new file mode 100644 index 0000000000..57f4cbe567 --- /dev/null +++ b/crates/ruborist/src/model/tarball_info.rs @@ -0,0 +1,55 @@ +//! Package tarball metadata for downloading and verification. + +use crate::model::compatibility::{is_cpu_compatible, is_os_compatible}; +use crate::model::manifest::VersionManifest; + +/// Package tarball information for downloading. +/// +/// A lightweight structure containing only the fields needed for +/// downloading and verifying a package tarball. Uses references to +/// avoid cloning data from the source manifest. +#[derive(Debug, Clone, Copy)] +pub struct PackageTarballInfo<'a> { + /// Package name + pub name: &'a str, + /// Resolved version + pub version: &'a str, + /// Tarball URL for downloading + pub tarball_url: Option<&'a str>, + /// Integrity hash for verification + pub integrity: Option<&'a str>, + /// OS compatibility constraint (if specified) + pub os: Option<&'a serde_json::Value>, + /// CPU compatibility constraint (if specified) + pub cpu: Option<&'a serde_json::Value>, +} + +impl PackageTarballInfo<'_> { + /// Check if this package is compatible with the current platform (os + cpu). + pub fn is_platform_compatible(&self) -> bool { + if let Some(os) = self.os + && !is_os_compatible(os) + { + return false; + } + if let Some(cpu) = self.cpu + && !is_cpu_compatible(cpu) + { + return false; + } + true + } +} + +impl<'a> From<&'a VersionManifest> for PackageTarballInfo<'a> { + fn from(m: &'a VersionManifest) -> Self { + Self { + name: &m.name, + version: &m.version, + tarball_url: m.dist.tarball.as_deref(), + integrity: m.dist.integrity.as_deref(), + os: m.os.as_ref(), + cpu: m.cpu.as_ref(), + } + } +} diff --git a/crates/ruborist/src/resolver/builder.rs b/crates/ruborist/src/resolver/builder.rs index 899a9f07ac..9489ea21b8 100644 --- a/crates/ruborist/src/resolver/builder.rs +++ b/crates/ruborist/src/resolver/builder.rs @@ -23,6 +23,7 @@ use std::collections::HashMap; use std::path::PathBuf; use crate::model::graph::{DependencyGraph, FindResult, PackageNode}; +use crate::model::manifest::NodeManifest; use crate::model::node::EdgeType; use crate::model::package_json::PackageJson; use crate::resolver::preload::{PreloadConfig, preload_manifests}; @@ -579,32 +580,48 @@ async fn run_bfs_phase( }); for edge_info in unresolved { + receiver.on_event(BuildEvent::Resolving { + name: &edge_info.name, + }); match process_dependency(graph, registry, node_index, &edge_info, legacy_peer_deps) .await? { ProcessResult::Created(idx) => { - receiver.on_event(BuildEvent::Resolved { - name: edge_info.name, - version: graph - .get_node(idx) - .map(|n| n.version.clone()) - .unwrap_or_default(), - }); + // Extract node info for events + if let Some(node) = graph.get_node(idx) { + receiver.on_event(BuildEvent::Resolved { + name: &edge_info.name, + version: &node.version, + }); + + // Send PackagePlaced for pipeline cloning + if let NodeManifest::Registry(ref manifest) = node.manifest { + // Get parent path for dependency ordering + let parent_path = graph + .get_node(node_index) + .map(|parent| parent.path.as_path()); + receiver.on_event(BuildEvent::PackagePlaced { + package: manifest.into(), + path: &node.path, + parent_path, + }); + } + } + next_level.push(idx); } ProcessResult::Reused(idx) => { - receiver.on_event(BuildEvent::Reused { - name: edge_info.name, - version: graph - .get_node(idx) - .map(|n| n.version.clone()) - .unwrap_or_default(), - }); + if let Some(node) = graph.get_node(idx) { + receiver.on_event(BuildEvent::Reused { + name: &edge_info.name, + version: &node.version, + }); + } } ProcessResult::Skipped => { receiver.on_event(BuildEvent::Skipped { - name: edge_info.name, - spec: edge_info.spec, + name: &edge_info.name, + spec: &edge_info.spec, }); } } diff --git a/crates/ruborist/src/resolver/preload.rs b/crates/ruborist/src/resolver/preload.rs index 092f18972f..7e009ba9d6 100644 --- a/crates/ruborist/src/resolver/preload.rs +++ b/crates/ruborist/src/resolver/preload.rs @@ -13,7 +13,7 @@ use crate::traits::progress::{BuildEvent, EventReceiver}; use crate::traits::registry::RegistryClient; /// Default concurrency limit for manifest fetching -pub const DEFAULT_CONCURRENCY: usize = 20; +pub const DEFAULT_CONCURRENCY: usize = 64; /// A dependency spec: (name, version_spec) pub type Dep = (String, String); @@ -159,7 +159,7 @@ where receiver.on_event(BuildEvent::PreloadQueued { count: 1 }); } - receiver.on_event(BuildEvent::PreloadFetching { name: name.clone() }); + receiver.on_event(BuildEvent::PreloadFetching { name: &name }); futures.push(async move { let start = tokio::time::Instant::now(); @@ -194,11 +194,14 @@ where tracing::debug!("Preloaded {}@{}", name, resolved.version); receiver.on_event(BuildEvent::PreloadProgress { - name: name.clone(), - version: resolved.version.clone(), + name: &name, + version: &resolved.version, current: stats.success_count, }); + // Send PackageResolved event for pipeline downloading + receiver.on_event(BuildEvent::PackageResolved((&resolved.manifest).into())); + pending.extend(extract_transitive_deps(&resolved.manifest, &config)); on_manifest(&name, resolved.manifest); } diff --git a/crates/ruborist/src/traits/progress.rs b/crates/ruborist/src/traits/progress.rs index bec66b6414..72c739a785 100644 --- a/crates/ruborist/src/traits/progress.rs +++ b/crates/ruborist/src/traits/progress.rs @@ -3,9 +3,13 @@ //! This module provides an event-driven approach for tracking dependency //! resolution progress without adding any I/O to ruborist itself. +use std::path::Path; + +pub use crate::model::tarball_info::PackageTarballInfo; + /// Events emitted during dependency resolution. -#[derive(Debug, Clone)] -pub enum BuildEvent { +#[derive(Debug, Clone, Copy)] +pub enum BuildEvent<'a> { /// Starting preload phase with N initial dependencies. PreloadStart { count: usize }, @@ -13,16 +17,21 @@ pub enum BuildEvent { PreloadQueued { count: usize }, /// A fetch task was started for a package. - PreloadFetching { name: String }, + PreloadFetching { name: &'a str }, /// A package was preloaded successfully. PreloadProgress { - name: String, - version: String, + name: &'a str, + version: &'a str, /// Current count of preloaded packages current: usize, }, + /// A package was fully resolved with download info. + /// This event enables pipeline downloading - tarball can be downloaded + /// immediately while other manifests are still being fetched. + PackageResolved(PackageTarballInfo<'a>), + /// Preload phase completed with success/failed counts. PreloadComplete { success: usize, failed: usize }, @@ -33,16 +42,27 @@ pub enum BuildEvent { DependencyCount { count: usize }, /// Starting to resolve a specific package. - Resolving { name: String }, + Resolving { name: &'a str }, /// Successfully resolved a package (reused existing). - Reused { name: String, version: String }, + Reused { name: &'a str, version: &'a str }, /// Successfully resolved a package (created new node). - Resolved { name: String, version: String }, + Resolved { name: &'a str, version: &'a str }, + + /// A package node was placed in the dependency tree. + /// This enables pipeline cloning - the package can be cloned + /// immediately after download, without waiting for full tree build. + PackagePlaced { + package: PackageTarballInfo<'a>, + /// Target path in node_modules (e.g., "node_modules/lodash") + path: &'a Path, + /// Parent package path (None for root-level dependencies) + parent_path: Option<&'a Path>, + }, /// Skipped an optional dependency. - Skipped { name: String, spec: String }, + Skipped { name: &'a str, spec: &'a str }, /// Completed a BFS level, next level has N nodes. LevelComplete { next_level_count: usize }, @@ -64,7 +84,7 @@ pub enum BuildEvent { /// struct MyReceiver; /// /// impl EventReceiver for MyReceiver { -/// fn on_event(&self, event: BuildEvent) { +/// fn on_event(&self, event: BuildEvent<'_>) { /// match event { /// BuildEvent::Resolving { name } => println!("Resolving {}...", name), /// BuildEvent::Resolved { name, version } => println!(" {}@{}", name, version), @@ -75,12 +95,12 @@ pub enum BuildEvent { /// ``` pub trait EventReceiver: Send + Sync { /// Called when a build event occurs. - fn on_event(&self, event: BuildEvent); + fn on_event(&self, event: BuildEvent<'_>); } /// A no-op event receiver for when event tracking is not needed. pub struct NoopReceiver; impl EventReceiver for NoopReceiver { - fn on_event(&self, _event: BuildEvent) {} + fn on_event(&self, _event: BuildEvent<'_>) {} }