diff --git a/.github/workflows/blocking-store-process-locks.yml b/.github/workflows/blocking-store-process-locks.yml new file mode 100644 index 00000000..aac5f964 --- /dev/null +++ b/.github/workflows/blocking-store-process-locks.yml @@ -0,0 +1,70 @@ +name: blocking store process locks + +on: + pull_request: + paths: + - src/store.rs + - src/install_graph.rs + - src/install_graph/** + - Cargo.toml + - Cargo.lock + - .github/workflows/blocking-store-process-locks.yml + push: + branches: [main] + paths: + - src/store.rs + - src/install_graph.rs + - src/install_graph/** + - Cargo.toml + - Cargo.lock + - .github/workflows/blocking-store-process-locks.yml + workflow_dispatch: + +permissions: + contents: read + +concurrency: + group: blocking-store-process-locks-${{ github.workflow }}-${{ github.event.pull_request.number || github.ref }} + cancel-in-progress: true + +env: + CARGO_TERM_COLOR: always + RUST_BACKTRACE: "1" + +jobs: + linux-kernel-lock: + name: Linux process-lock and recursive-cache contracts + runs-on: ubuntu-24.04 + timeout-minutes: 35 + steps: + - name: Check out zed-cli + uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1 + with: + fetch-depth: 1 + persist-credentials: false + show-progress: false + + - name: Install stable Rust with formatting and lint support + run: | + rustup toolchain install stable --profile minimal --component rustfmt,clippy + rustup default stable + rustc --version --verbose + cargo --version --verbose + + - name: Check formatting + run: cargo fmt --all --check + + - name: Prove a separate process blocks until orderly release + run: cargo test --locked --lib store::tests::install_process_lock_blocks_without_polling_until_owner_release -- --exact --nocapture + + - name: Prove owner termination releases the descriptor lock + run: cargo test --locked --lib store::tests::install_process_lock_is_released_when_owner_process_is_terminated -- --exact --nocapture + + - name: Run all store process-lock tests + run: cargo test --locked --lib store::tests + + - name: Run recursive acquisition and cache-race tests + run: cargo test --locked --lib install_graph::tests + + - name: Deny Rust warnings + run: cargo clippy --locked --all-targets -- -D warnings diff --git a/.github/workflows/recursive-install-windows.yml b/.github/workflows/recursive-install-windows.yml new file mode 100644 index 00000000..eda09e0f --- /dev/null +++ b/.github/workflows/recursive-install-windows.yml @@ -0,0 +1,87 @@ +name: recursive install windows locking + +on: + pull_request: + paths: + - src/install_graph.rs + - src/install_graph/** + - src/store.rs + - Cargo.toml + - Cargo.lock + - .github/workflows/recursive-install-windows.yml + push: + branches: [main] + paths: + - src/install_graph.rs + - src/install_graph/** + - src/store.rs + - Cargo.toml + - Cargo.lock + - .github/workflows/recursive-install-windows.yml + workflow_dispatch: + +permissions: + contents: read + +concurrency: + group: recursive-install-windows-${{ github.workflow }}-${{ github.event.pull_request.number || github.ref }} + cancel-in-progress: true + +env: + CARGO_TERM_COLOR: always + RUST_BACKTRACE: "1" + +jobs: + windows-kernel-lock: + name: Windows process lock and shared-home races + runs-on: windows-latest + timeout-minutes: 35 + defaults: + run: + shell: pwsh + steps: + - name: Check out zed-cli + uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1 + with: + path: zed-cli + fetch-depth: 1 + persist-credentials: false + show-progress: false + + - name: Check out pinned zed-interfaces sibling + uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1 + with: + repository: zed-pkg/zed-interfaces + ref: dc0e0a0620b9462817950b552d3d334a184b1cb1 + path: zed-interfaces + fetch-depth: 1 + persist-credentials: false + show-progress: false + + - name: Install stable Rust with formatting support + run: | + rustup toolchain install stable --profile minimal --component rustfmt + rustup default stable + rustc --version --verbose + cargo --version --verbose + + - name: Check formatting + run: cargo fmt --all --check --manifest-path zed-cli/Cargo.toml + + - name: Prove a separate process blocks on the same artifact lock + run: cargo test --locked --manifest-path zed-cli/Cargo.toml --lib install_graph::tests::artifact_waiters_block_until_the_owner_releases_the_kernel_lock -- --exact --nocapture + + - name: Prove concurrent prefetches deduplicate a shared artifact + run: cargo test --locked --manifest-path zed-cli/Cargo.toml --lib install_graph::tests::concurrent_prefetches_share_one_artifact_download -- --exact --nocapture + + - name: Prove recursive and transactional paths share one lock + run: cargo test --locked --manifest-path zed-cli/Cargo.toml --lib install_graph::tests::legacy_and_recursive_acquisition_paths_share_one_download_lock -- --exact --nocapture + + - name: Prove failed downloads leave no cache publication + run: cargo test --locked --manifest-path zed-cli/Cargo.toml --lib install_graph::tests::failed_downloads_remove_staging_files_and_never_publish_cache_entries -- --exact --nocapture + + - name: Prove the lower-level store lock blocks across processes + run: cargo test --locked --manifest-path zed-cli/Cargo.toml --lib store::tests::install_process_lock_blocks_without_polling_until_owner_release -- --exact --nocapture + + - name: Prove owner termination releases the lower-level store lock + run: cargo test --locked --manifest-path zed-cli/Cargo.toml --lib store::tests::install_process_lock_is_released_when_owner_process_is_terminated -- --exact --nocapture diff --git a/docs/recursive-installs.md b/docs/recursive-installs.md new file mode 100644 index 00000000..56376fa6 --- /dev/null +++ b/docs/recursive-installs.md @@ -0,0 +1,89 @@ +# Recursive and concurrent installs + +`zed install` resolves dependencies from every installed package manifest, not +only the consumer's direct dependency table. The resolver keeps one selected +version per `org/name`, detects incompatible requirements, terminates cycles, +and deduplicates diamond-shaped graphs before the project tree is changed. + +## Concurrency + +Artifact download, integrity verification, and extraction run through a bounded +worker queue. The default is five concurrent artifact installs: + +```text +ZED_PKG_INSTALL_CONCURRENCY=5 zed install +``` + +`ZED_PKG_INSTALL_CONCURRENCY` may be set from 1 through 32. Invalid values and +zero use the default. Registry metadata resolution stays deterministic on the +coordinator thread; only independent artifact work is concurrent. + +## Cross-process locking + +Each content hash owns a lock file under: + +```text +$ZED_PKG_HOME/locks/artifact-.lock +``` + +The lock uses the operating system's blocking file-lock primitive. A waiter +sleeps in the kernel until the owner exits or releases the descriptor; there is +no retry timer or filesystem polling loop. The lock covers download, SHA-256 +verification, atomic publication into the download cache, and extraction into +the content-addressed store. Two terminals or CI jobs sharing a Zed home can +therefore install unrelated artifacts concurrently while downloading any one +artifact at most once. + +Recursive prefetch and the transactional installer use the same artifact +acquisition function. Frozen installs, build preparation, `zed add`, and +`zed remove` therefore cannot bypass the per-hash lock through an older direct +cache-download path. + +Downloads are written to a temporary file in the cache directory, verified, +and atomically renamed. A corrupt cache file left by an interrupted older +client is removed and fetched again while the artifact lock is held. + +## Materialization + +The concurrent phase only warms the global store. The existing project +transaction remains responsible for lockfile writes, adapter wiring, rollback, +and dependency materialization. Unix installations therefore continue to use +store-backed symlinks by default; explicit copy mode and the non-Unix fallback +remain self-contained copies. + +## Frozen and workspace compatibility + +Recursive acquisition is composed with the strict frozen-lock boundary. A +frozen install may verify artifacts, repair deleted links, reconstruct workspace +projections, and refresh generated adapter wiring, but it preserves the caller's +committed lockfile bytes exactly. It does not rewrite comments, provenance, or +fields introduced by a newer CLI. + +Workspace packages still enter the ordinary adapter and transaction lifecycle +through the `ops_entry` dispatcher. This keeps direct and transitive workspace +links, copy mode, uninstall cleanup, and post-uninstall frozen restoration on +the same fail-closed path as registry packages. + +The recursive install modules coexist with the resolver-only frozen fetch path, +Nix fixed-output bundle generation, canonical export planning, and standalone +pure flake rendering. Those export surfaces consume the same validated lock and +content identities; they do not bypass recursive integrity checks or per-hash +locking. + +## Validation + +The unit suite races the recursive and transactional acquisition paths against +the same absent content hash. The first download is held open while the second +path attempts acquisition; the test proves that the second path never enters +its registry download and that the aggregate download count remains one. + +A separate parent/child-process regression holds an artifact lock in the parent, +proves that the child remains blocked after reaching acquisition, releases the +owner descriptor, and then requires the child to wake and acquire successfully. +The same contract runs on Windows together with the shared-home deduplication, +mixed recursive/transactional acquisition, and failed-download cleanup tests. + +The companion end-to-end suite also runs multiple independent CLI processes +against one shared Zed home, verifies the complete transitive lock graph, and +checks that project packages are store-backed symlinks rather than copied +package trees. diff --git a/src/config.rs b/src/config.rs index 34bf3811..89210180 100644 --- a/src/config.rs +++ b/src/config.rs @@ -156,6 +156,7 @@ struct ManifestOverride { thread_local! { static MANIFEST_OVERRIDE: RefCell> = const { RefCell::new(None) }; + static INSTALL_PREFETCH_CONFIG: RefCell> = const { RefCell::new(None) }; } struct ManifestOverrideGuard; @@ -168,10 +169,44 @@ impl Drop for ManifestOverrideGuard { } } +struct InstallPrefetchGuard; + +impl Drop for InstallPrefetchGuard { + fn drop(&mut self) { + INSTALL_PREFETCH_CONFIG.with(|slot| { + slot.borrow_mut().take(); + }); + } +} + fn normalized_project(project: &Path) -> PathBuf { fs::canonicalize(project).unwrap_or_else(|_| project.to_path_buf()) } +/// Mark one facade operation so an in-memory manifest proposed by `zed add` +/// or `zed remove` is recursively prefetched before the implementation-local +/// reinstall begins. The context is thread-local and panic-safe; ordinary +/// manifest overrides remain side-effect free. +pub(crate) fn with_install_prefetch( + cfg: &Config, + operation: impl FnOnce() -> Result, +) -> Result { + INSTALL_PREFETCH_CONFIG.with(|slot| { + let mut slot = slot.borrow_mut(); + if slot.is_some() { + return Err(anyhow!( + "an install prefetch context is already active on this thread" + )); + } + *slot = Some(cfg.clone()); + Ok(()) + })?; + let guard = InstallPrefetchGuard; + let result = operation(); + drop(guard); + result +} + /// Run one installer operation with an in-memory root manifest. Package /// manifests read from the store still come from disk; only the exact consumer /// directory is overridden. The guard is thread-local and panic-safe, so a @@ -189,10 +224,17 @@ pub(crate) fn with_manifest_override( "a manifest override is already active on this thread" )); } - *slot = Some(ManifestOverride { project, text }); + *slot = Some(ManifestOverride { + project: project.clone(), + text, + }); Ok(()) })?; let guard = ManifestOverrideGuard; + let prefetch_cfg = INSTALL_PREFETCH_CONFIG.with(|slot| slot.borrow().clone()); + if let Some(cfg) = prefetch_cfg { + crate::install_graph::prefetch(&project, &cfg, false)?; + } let result = operation(); drop(guard); result @@ -282,6 +324,32 @@ url = "https://localhost/manifestless/consumer" assert!(error.to_string().contains("already active")); } + #[test] + fn install_prefetch_context_is_scoped_and_nested_contexts_fail_closed() { + let home = tempfile::tempdir().unwrap(); + let cfg = Config { + registry: "file:///unused".to_string(), + home: home.path().to_path_buf(), + token: None, + auth_url: "https://localhost/unused".to_string(), + supabase_url: None, + supabase_key: None, + interactive: false, + }; + + with_install_prefetch(&cfg, || { + let active = INSTALL_PREFETCH_CONFIG.with(|slot| slot.borrow().is_some()); + assert!(active); + let nested = with_install_prefetch(&cfg, || Ok(())).unwrap_err(); + assert!(nested.to_string().contains("already active")); + Ok(()) + }) + .unwrap(); + + let active = INSTALL_PREFETCH_CONFIG.with(|slot| slot.borrow().is_some()); + assert!(!active); + } + #[test] fn relative_home_is_resolved_from_the_invocation_directory() { let globals = Globals { diff --git a/src/install_graph.rs b/src/install_graph.rs new file mode 100644 index 00000000..59452c14 --- /dev/null +++ b/src/install_graph.rs @@ -0,0 +1,303 @@ +//! Recursive, bounded-concurrency artifact prefetch for `zed install`. +//! +//! The normal installer remains the authority for resolution, lockfile writes, +//! project transactions, adapters, and symlink/copy materialization. This module +//! runs immediately before it: it walks package manifests recursively, installs +//! artifacts into the content-addressed store with a five-worker queue, and lets +//! the normal install phase reuse the warm store. +//! +//! Cross-process coordination is deliberately per artifact. Waiters block in +//! the operating system's file-lock primitive instead of polling, so unrelated +//! dependencies and unrelated projects can keep making progress. + +use std::collections::{BTreeMap, BTreeSet, VecDeque}; +use std::fs; +use std::path::{Path, PathBuf}; +use std::sync::{Arc, Condvar, Mutex, MutexGuard, mpsc}; +use std::thread::{self, JoinHandle}; + +use anyhow::{Context, Result, bail}; +use fs2::FileExt; +use globset::Glob; +use zed_interfaces::lockfile::Lockfile; +use zed_interfaces::manifest::{Manifest, is_slug}; +use zed_interfaces::paths::{LOCKFILE_FILE, MANIFEST_FILE}; +use zed_interfaces::registry::VersionMetadata; +use zed_interfaces::version::{self, Requirement}; + +use crate::config::{Config, read_manifest}; +use crate::pack::sha256_file; +use crate::registry::{Registry, registry_for}; +use crate::store::{Store, require_sha256}; + +/// Default number of artifact installs allowed to make progress at once. +pub const DEFAULT_INSTALL_CONCURRENCY: usize = 5; +const MAX_INSTALL_CONCURRENCY: usize = 32; + +#[derive(Debug, Default, Clone, Copy, PartialEq, Eq)] +pub struct PrefetchReport { + /// Unique registry packages resolved (workspace links are not counted). + pub resolved: usize, + /// Artifacts that were absent locally and downloaded by this process. + pub downloaded: usize, +} + +#[derive(Debug, Clone)] +struct DependencyRequest { + key: String, + requirement: String, +} + +#[derive(Debug, Clone)] +struct FetchTask { + sequence: usize, + key: String, + version: VersionMetadata, +} + +#[derive(Debug)] +struct FetchResult { + dependencies: BTreeMap, + downloaded: bool, +} + +#[derive(Debug)] +struct FetchMessage { + sequence: usize, + result: Result, +} + +#[derive(Debug, Default)] +struct QueueState { + tasks: VecDeque, + closed: bool, +} + +#[derive(Debug, Default)] +struct TaskQueue { + state: Mutex, + ready: Condvar, +} + +impl TaskQueue { + fn state(&self) -> MutexGuard<'_, QueueState> { + self.state + .lock() + .unwrap_or_else(|poisoned| poisoned.into_inner()) + } + + fn push(&self, task: FetchTask) -> bool { + let mut state = self.state(); + if state.closed { + return false; + } + state.tasks.push_back(task); + self.ready.notify_one(); + true + } + + fn pop(&self) -> Option { + let mut state = self.state(); + loop { + if let Some(task) = state.tasks.pop_front() { + return Some(task); + } + if state.closed { + return None; + } + state = self + .ready + .wait(state) + .unwrap_or_else(|poisoned| poisoned.into_inner()); + } + } + + fn close(&self) { + let mut state = self.state(); + state.closed = true; + self.ready.notify_all(); + } + + fn cancel(&self) { + let mut state = self.state(); + state.tasks.clear(); + state.closed = true; + self.ready.notify_all(); + } +} + +struct FetchPool { + queue: Arc, + results: mpsc::Receiver, + workers: Vec>, +} + +impl FetchPool { + fn new(concurrency: usize, registry_url: &str, home: &Path) -> Result { + let queue = Arc::new(TaskQueue::default()); + let (result_tx, results) = mpsc::channel(); + let mut workers = Vec::with_capacity(concurrency); + + for index in 0..concurrency { + let worker_queue = Arc::clone(&queue); + let worker_tx = result_tx.clone(); + let worker_registry = registry_url.to_string(); + let worker_home = home.to_path_buf(); + let spawn = thread::Builder::new() + .name(format!("zed-install-{index}")) + .spawn(move || worker_loop(worker_queue, worker_tx, worker_registry, worker_home)); + match spawn { + Ok(worker) => workers.push(worker), + Err(error) => { + queue.cancel(); + for worker in workers { + let _ = worker.join(); + } + return Err(error).context("starting recursive install worker"); + } + } + } + drop(result_tx); + Ok(Self { + queue, + results, + workers, + }) + } + + fn submit(&self, task: FetchTask) -> Result<()> { + if !self.queue.push(task) { + bail!("recursive install worker queue closed unexpectedly"); + } + Ok(()) + } + + fn receive(&self) -> Result { + self.results + .recv() + .context("all recursive install workers stopped unexpectedly") + } + + fn shutdown(self, cancel: bool) -> Result<()> { + if cancel { + self.queue.cancel(); + } else { + self.queue.close(); + } + let mut panicked = false; + for worker in self.workers { + panicked |= worker.join().is_err(); + } + if panicked { + bail!("a recursive install worker panicked"); + } + Ok(()) + } +} + +/// A process lock for one content hash. `lock_exclusive` blocks in the kernel; +/// it does not wake up periodically to retry. The file descriptor owns the +/// lock, so process exit or panic releases it automatically. +struct ArtifactProcessLock { + file: fs::File, +} + +impl ArtifactProcessLock { + fn acquire(home: &Path, sha256: &str) -> Result { + require_sha256(sha256)?; + let path = home.join("locks").join(format!("artifact-{sha256}.lock")); + if let Some(parent) = path.parent() { + fs::create_dir_all(parent)?; + } + let file = fs::OpenOptions::new() + .create(true) + .read(true) + .write(true) + .truncate(false) + .open(&path) + .with_context(|| format!("opening artifact lock {}", path.display()))?; + FileExt::lock_exclusive(&file) + .with_context(|| format!("waiting for artifact lock {}", path.display()))?; + Ok(Self { file }) + } +} + +impl Drop for ArtifactProcessLock { + fn drop(&mut self) { + let _ = FileExt::unlock(&self.file); + } +} + +#[derive(Debug, Default)] +struct WorkspaceGraph { + dependencies: BTreeMap>, +} + +impl WorkspaceGraph { + fn discover(project: &Path) -> Self { + let mut current = Some(project); + while let Some(directory) = current { + if directory.join(MANIFEST_FILE).is_file() + && let Ok(manifest) = read_manifest(directory) + && let Some(workspace) = manifest.workspace.as_ref() + { + return Self::collect(directory, &workspace.members); + } + current = directory.parent(); + } + Self::default() + } + + fn collect(root: &Path, patterns: &[String]) -> Self { + let mut graph = Self::default(); + for pattern in patterns { + let mut candidates = vec![root.to_path_buf()]; + for segment in pattern.split('/') { + let mut next = Vec::new(); + for base in &candidates { + if segment.contains('*') { + let Ok(glob) = Glob::new(segment) else { + continue; + }; + let matcher = glob.compile_matcher(); + if let Ok(entries) = fs::read_dir(base) { + for entry in entries.flatten() { + let name = entry.file_name(); + if entry.path().is_dir() + && matcher.is_match(Path::new(&name)) + && !name.to_string_lossy().starts_with('.') + { + next.push(entry.path()); + } + } + } + } else { + let candidate = base.join(segment); + if candidate.is_dir() { + next.push(candidate); + } + } + } + candidates = next; + } + for member_dir in candidates { + if let Ok(member) = read_manifest(&member_dir) { + let key = member.full_name(); + graph.dependencies.insert(key, member.dependencies); + } + } + } + graph + } +} + +mod artifact; +#[cfg(test)] +mod hardening_tests; +mod resolver; +#[cfg(test)] +mod tests; + +pub(crate) use artifact::ensure_artifact; +use artifact::worker_loop; +pub use resolver::prefetch; diff --git a/src/install_graph/artifact.rs b/src/install_graph/artifact.rs new file mode 100644 index 00000000..fe4b051e --- /dev/null +++ b/src/install_graph/artifact.rs @@ -0,0 +1,258 @@ +use super::*; + +pub(super) fn worker_loop( + queue: Arc, + results: mpsc::Sender, + registry_url: String, + home: PathBuf, +) { + let store = Store::new(&home); + let mut registry: Option> = None; + + while let Some(task) = queue.pop() { + let sequence = task.sequence; + let key = task.key.clone(); + let message = run_fetch_task(sequence, &key, || -> Result { + if registry.is_none() { + registry = Some(registry_for(®istry_url)?); + } + let registry = registry + .as_deref() + .context("recursive install worker has no registry")?; + prefetch_one(registry, &store, task) + }); + if results.send(message).is_err() { + return; + } + } +} + +/// Convert a task panic into the same sequenced result channel used for normal +/// failures. Without this boundary, the worker would unwind after popping the +/// task, permanently losing that sequence while the coordinator waited for a +/// result that could never arrive. +pub(super) fn run_fetch_task(sequence: usize, key: &str, work: F) -> FetchMessage +where + F: FnOnce() -> Result, +{ + let result = + std::panic::catch_unwind(std::panic::AssertUnwindSafe(work)).unwrap_or_else(|payload| { + let detail = payload + .downcast_ref::<&str>() + .copied() + .or_else(|| payload.downcast_ref::().map(String::as_str)) + .unwrap_or("non-string panic payload"); + Err(anyhow::anyhow!( + "recursive install worker panicked while processing {} (sequence {}): {}", + key, + sequence, + detail + )) + }); + FetchMessage { sequence, result } +} + +pub(super) fn prefetch_one( + registry: &dyn Registry, + store: &Store, + task: FetchTask, +) -> Result { + let (package_dir, downloaded) = ensure_artifact(registry, store, &task.version) + .with_context(|| format!("prefetching {}@{}", task.key, task.version.version))?; + let manifest_path = package_dir.join(MANIFEST_FILE); + let dependencies = if manifest_path.is_file() { + let manifest = read_manifest(&package_dir).with_context(|| { + format!( + "reading dependency manifest for {}@{} from {}", + task.key, + task.version.version, + manifest_path.display() + ) + })?; + if manifest.full_name() != task.key || manifest.package.version != task.version.version { + bail!( + "artifact manifest declares {}@{} while registry metadata selected {}@{}; refusing transitive dependency expansion", + manifest.full_name(), + manifest.package.version, + task.key, + task.version.version + ); + } + manifest.dependencies + } else { + // Keep legacy manifestless artifacts installable as leaves, but never + // hide a malformed or identity-mismatched manifest when one is present. + BTreeMap::new() + }; + Ok(FetchResult { + dependencies, + downloaded, + }) +} + +/// Acquire one immutable artifact through the shared cache/store path. +/// +/// Every caller uses the same blocking per-hash process lock, staged download, +/// integrity check, atomic cache publication, and extraction sequence. This is +/// intentionally crate-visible so the legacy transactional installer cannot +/// bypass recursive-prefetch locking. +pub(crate) fn ensure_artifact( + registry: &dyn Registry, + store: &Store, + version: &VersionMetadata, +) -> Result<(PathBuf, bool)> { + validate_version_identity(version, &version.org, &version.name, &version.version)?; + if store.has(&version.sha256) { + return Ok((store.pkg_dir(&version.sha256), false)); + } + + let _artifact_lock = ArtifactProcessLock::acquire(store.home(), &version.sha256)?; + if store.has(&version.sha256) { + return Ok((store.pkg_dir(&version.sha256), false)); + } + + let cached = store.cached_artifact(&version.sha256); + let mut downloaded = false; + if !cached.is_file() { + download_atomic(registry, version, &cached)?; + downloaded = true; + } + + match store.add_artifact(&cached, &version.sha256) { + Ok(package_dir) => Ok((package_dir, downloaded)), + Err(first_error) if !downloaded => { + // A killed legacy client may have left a partial cache file. The + // per-artifact lock makes removal and replacement safe. + let _ = fs::remove_file(&cached); + download_atomic(registry, version, &cached)?; + store + .add_artifact(&cached, &version.sha256) + .with_context(|| { + format!("cached artifact was invalid ({first_error:#}); redownload also failed") + }) + .map(|package_dir| (package_dir, true)) + } + Err(error) => Err(error), + } +} + +fn download_atomic( + registry: &dyn Registry, + version: &VersionMetadata, + cached: &Path, +) -> Result<()> { + let cache_dir = cached.parent().context("cached artifact has a parent")?; + fs::create_dir_all(cache_dir)?; + let staging_dir = tempfile::tempdir_in(cache_dir)?; + let staged = staging_dir.path().join("artifact.download"); + registry.download(version, &staged)?; + + let (actual_sha256, _) = sha256_file(&staged)?; + if actual_sha256 != version.sha256 { + bail!( + "artifact hash mismatch: expected {}, got {}", + version.sha256, + actual_sha256 + ); + } + + if cached.exists() { + let (cached_sha256, _) = sha256_file(cached)?; + if cached_sha256 == version.sha256 { + return Ok(()); + } + fs::remove_file(cached)?; + } + match fs::rename(&staged, cached) { + Ok(()) => Ok(()), + Err(error) if cached.is_file() => { + let (cached_sha256, _) = sha256_file(cached)?; + if cached_sha256 == version.sha256 { + Ok(()) + } else { + Err(error).with_context(|| { + format!("publishing downloaded artifact to {}", cached.display()) + }) + } + } + Err(error) => Err(error) + .with_context(|| format!("publishing downloaded artifact to {}", cached.display())), + } +} + +pub(super) fn split_key(key: &str) -> Result<(String, String)> { + let mut parts = key.splitn(2, '/'); + match (parts.next(), parts.next()) { + (Some(org), Some(name)) if is_slug(org) && is_slug(name) => { + Ok((org.to_string(), name.to_string())) + } + _ => bail!( + "invalid package spec `{key}` (expected slug/slug without path traversal or extra segments)" + ), + } +} + +pub(super) fn validate_version_identity( + version: &VersionMetadata, + expected_org: &str, + expected_name: &str, + expected_version: &str, +) -> Result<()> { + if !is_slug(&version.org) || !is_slug(&version.name) { + bail!( + "registry returned invalid package identity `{}/{}`; refusing", + version.org, + version.name + ); + } + require_sha256(&version.sha256)?; + if version.org != expected_org + || version.name != expected_name + || version.version != expected_version + { + bail!( + "registry returned `{}/{}@{}` while resolving `{expected_org}/{expected_name}@{expected_version}`; refusing", + version.org, + version.name, + version.version + ); + } + Ok(()) +} + +pub(super) fn resolve_version( + registry: &dyn Registry, + org: &str, + name: &str, + requirement_text: &str, +) -> Result { + let key = format!("{org}/{name}"); + let requirement = Requirement::parse(requirement_text); + let package = registry.get_package(org, name)?; + let mut candidates = package.versions.clone(); + let mut skipped_yanked = Vec::new(); + + loop { + let selected = version::resolve(&requirement, &candidates).map(str::to_string); + let Some(selected) = selected else { + if !skipped_yanked.is_empty() { + bail!( + "the only version(s) of {key} satisfying `{requirement_text}` are yanked ({}); existing lockfiles keep working via `zed install --frozen`", + skipped_yanked.join(", ") + ); + } + bail!( + "no version of {key} satisfies `{requirement_text}` (available: {})", + package.versions.join(", ") + ); + }; + let version = registry.get_version(org, name, &selected)?; + validate_version_identity(&version, org, name, &selected)?; + if version.yanked { + candidates.retain(|candidate| *candidate != selected); + skipped_yanked.push(selected); + continue; + } + return Ok(version); + } +} diff --git a/src/install_graph/hardening_tests.rs b/src/install_graph/hardening_tests.rs new file mode 100644 index 00000000..0c248cdd --- /dev/null +++ b/src/install_graph/hardening_tests.rs @@ -0,0 +1,155 @@ +use anyhow::anyhow; + +use super::*; +use crate::pack::pack; + +fn manifest_text(org: &str, name: &str, version: &str) -> String { + format!( + r#"[package] +org = "{org}" +name = "{name}" +version = "{version}" + +[package.repository] +vcs = "git" +url = "https://example.invalid/{org}/{name}" +"#, + ) +} + +fn prepared_artifact( + temp: &tempfile::TempDir, + stored_manifest: Option<&str>, +) -> (Store, FetchTask, Box) { + let source = temp.path().join("source"); + fs::create_dir_all(&source).unwrap(); + let selected_manifest = manifest_text("test", "selected", "1.0.0"); + fs::write(source.join(MANIFEST_FILE), &selected_manifest).unwrap(); + fs::write(source.join("payload.txt"), b"fixture\n").unwrap(); + let manifest = Manifest::parse(&selected_manifest).unwrap(); + let packed = pack(&source, &manifest, Some(&temp.path().join("packed"))).unwrap(); + + let store = Store::new(&temp.path().join("home")); + let package_dir = store.add_artifact(&packed.path, &packed.sha256).unwrap(); + match stored_manifest { + Some(text) => fs::write(package_dir.join(MANIFEST_FILE), text).unwrap(), + None => fs::remove_file(package_dir.join(MANIFEST_FILE)).unwrap(), + } + + let task = FetchTask { + sequence: 0, + key: "test/selected".to_string(), + version: VersionMetadata { + org: "test".to_string(), + name: "selected".to_string(), + version: "1.0.0".to_string(), + sha256: packed.sha256, + size: packed.size, + format: packed.format, + vcs_tag: "v1.0.0".to_string(), + vcs_commit: Some("0123456789abcdef0123456789abcdef01234567".to_string()), + download_url: "file:///unused-artifact.tar.gz".to_string(), + published_at: "1970-01-01T00:00:00Z".to_string(), + yanked: false, + }, + }; + + let registry_root = temp.path().join("registry"); + fs::create_dir_all(®istry_root).unwrap(); + let registry = registry_for(&format!("file://{}", registry_root.display())).unwrap(); + (store, task, registry) +} + +#[test] +fn dependency_keys_are_slug_validated_before_registry_access() { + for key in [ + "../escape", + "test/../escape", + "test/name/extra", + "/name", + "org/", + ] { + let error = super::artifact::split_key(key).unwrap_err().to_string(); + assert!(error.contains("invalid package spec"), "{key}: {error}"); + } + + assert_eq!( + super::artifact::split_key("valid-org/valid-name").unwrap(), + ("valid-org".to_string(), "valid-name".to_string()) + ); +} + +#[test] +fn malformed_embedded_manifests_fail_closed_instead_of_truncating_the_graph() { + let temp = tempfile::tempdir().unwrap(); + let (store, task, registry) = prepared_artifact(&temp, Some("[package\ninvalid = true\n")); + + let error = super::artifact::prefetch_one(registry.as_ref(), &store, task).unwrap_err(); + let message = format!("{error:#}"); + assert!(message.contains("reading dependency manifest"), "{message}"); + assert!(message.contains("test/selected@1.0.0"), "{message}"); +} + +#[test] +fn embedded_manifest_identity_must_match_selected_registry_metadata() { + let temp = tempfile::tempdir().unwrap(); + let mismatched = manifest_text("test", "different", "2.0.0"); + let (store, task, registry) = prepared_artifact(&temp, Some(&mismatched)); + + let error = super::artifact::prefetch_one(registry.as_ref(), &store, task) + .unwrap_err() + .to_string(); + assert!( + error.contains("artifact manifest declares test/different@2.0.0"), + "{error}" + ); + assert!(error.contains("test/selected@1.0.0"), "{error}"); +} + +#[test] +fn legacy_manifestless_artifacts_remain_valid_leaf_packages() { + let temp = tempfile::tempdir().unwrap(); + let (store, task, registry) = prepared_artifact(&temp, None); + + let result = super::artifact::prefetch_one(registry.as_ref(), &store, task).unwrap(); + assert!(result.dependencies.is_empty()); + assert!(!result.downloaded); +} + +#[test] +fn sequenced_receive_buffers_later_failures_until_earlier_results_arrive() { + let queue = Arc::new(TaskQueue::default()); + let (sender, results) = mpsc::channel(); + let pool = FetchPool { + queue, + results, + workers: Vec::new(), + }; + + sender + .send(FetchMessage { + sequence: 1, + result: Err(anyhow!( + "later lockfile entry failed first in wall-clock time" + )), + }) + .unwrap(); + sender + .send(FetchMessage { + sequence: 0, + result: Ok(FetchResult { + dependencies: BTreeMap::new(), + downloaded: false, + }), + }) + .unwrap(); + drop(sender); + + let mut buffered = BTreeMap::new(); + let first = super::resolver::receive_in_order(&pool, &mut buffered, 0).unwrap(); + assert!(first.dependencies.is_empty()); + let error = super::resolver::receive_in_order(&pool, &mut buffered, 1) + .unwrap_err() + .to_string(); + assert!(error.contains("later lockfile entry failed"), "{error}"); +} diff --git a/src/install_graph/resolver.rs b/src/install_graph/resolver.rs new file mode 100644 index 00000000..dd41abd4 --- /dev/null +++ b/src/install_graph/resolver.rs @@ -0,0 +1,231 @@ +use super::artifact::{resolve_version, split_key, validate_version_identity}; +use super::*; + +/// Warm the content-addressed store for an install. The normal installer runs +/// immediately afterward and remains responsible for the project transaction. +pub fn prefetch(project: &Path, cfg: &Config, frozen: bool) -> Result { + let concurrency = install_concurrency(); + let registry = registry_for(&cfg.registry)?; + let report = if frozen { + prefetch_locked(project, cfg, registry.as_ref(), concurrency)? + } else { + let manifest = read_manifest(project)?; + prefetch_recursive(project, cfg, registry.as_ref(), &manifest, concurrency)? + }; + + if report.resolved > 0 { + eprintln!( + "recursive install prefetch: {} package(s), up to {} concurrent, {} downloaded", + report.resolved, concurrency, report.downloaded + ); + } + Ok(report) +} + +fn install_concurrency() -> usize { + normalize_concurrency(std::env::var("ZED_PKG_INSTALL_CONCURRENCY").ok().as_deref()) +} + +pub(super) fn normalize_concurrency(raw: Option<&str>) -> usize { + raw.and_then(|value| value.parse::().ok()) + .filter(|value| *value > 0) + .unwrap_or(DEFAULT_INSTALL_CONCURRENCY) + .min(MAX_INSTALL_CONCURRENCY) +} + +fn run_with_pool( + cfg: &Config, + concurrency: usize, + operation: impl FnOnce(&FetchPool) -> Result, +) -> Result { + let pool = FetchPool::new(concurrency, &cfg.registry, &cfg.home)?; + match operation(&pool) { + Ok(value) => { + pool.shutdown(false)?; + Ok(value) + } + Err(error) => { + let _ = pool.shutdown(true); + Err(error) + } + } +} + +fn prefetch_locked( + project: &Path, + cfg: &Config, + registry: &dyn Registry, + concurrency: usize, +) -> Result { + let lock_path = project.join(LOCKFILE_FILE); + let text = fs::read_to_string(&lock_path) + .with_context(|| format!("--frozen requires {}", lock_path.display()))?; + let lock = Lockfile::parse(&text)?; + let mut tasks = Vec::with_capacity(lock.packages.len()); + let mut seen = BTreeSet::new(); + + for locked in &lock.packages { + if !is_slug(&locked.org) || !is_slug(&locked.name) { + bail!( + "lockfile entry `{}/{}` has an invalid identity; refusing", + locked.org, + locked.name + ); + } + require_sha256(&locked.sha256)?; + let key = locked.full_name(); + if !seen.insert(key.clone()) { + bail!("duplicate package `{key}` in {LOCKFILE_FILE}"); + } + let version = registry.get_version(&locked.org, &locked.name, &locked.version)?; + validate_version_identity(&version, &locked.org, &locked.name, &locked.version)?; + if version.sha256 != locked.sha256 { + bail!( + "registry artifact for {}@{} changed (lock {} vs registry {}); refusing", + key, + locked.version, + locked.sha256, + version.sha256 + ); + } + tasks.push(FetchTask { + sequence: tasks.len(), + key, + version, + }); + } + + if tasks.is_empty() { + return Ok(PrefetchReport::default()); + } + + run_with_pool(cfg, concurrency, |pool| { + let total = tasks.len(); + for task in tasks { + pool.submit(task)?; + } + let mut downloaded = 0usize; + let mut buffered_results: BTreeMap> = BTreeMap::new(); + for expected in 0..total { + let result = receive_in_order(pool, &mut buffered_results, expected)?; + downloaded += usize::from(result.downloaded); + } + Ok(PrefetchReport { + resolved: total, + downloaded, + }) + }) +} + +fn prefetch_recursive( + project: &Path, + cfg: &Config, + registry: &dyn Registry, + manifest: &Manifest, + concurrency: usize, +) -> Result { + if manifest.dependencies.is_empty() { + return Ok(PrefetchReport::default()); + } + let workspace = WorkspaceGraph::discover(project); + + run_with_pool(cfg, concurrency, |pool| { + let mut pending: VecDeque = manifest + .dependencies + .iter() + .map(|(key, requirement)| DependencyRequest { + key: key.clone(), + requirement: requirement.clone(), + }) + .collect(); + let mut resolved: BTreeMap = BTreeMap::new(); + let mut expanded_workspace = BTreeSet::new(); + let mut in_flight = 0usize; + let mut next_sequence = 0usize; + let mut next_result = 0usize; + let mut buffered_results: BTreeMap> = BTreeMap::new(); + let mut downloaded = 0usize; + + loop { + while let Some(request) = pending.pop_front() { + if let Some(member_dependencies) = workspace.dependencies.get(&request.key) { + if expanded_workspace.insert(request.key.clone()) { + pending.extend(member_dependencies.iter().map(|(key, requirement)| { + DependencyRequest { + key: key.clone(), + requirement: requirement.clone(), + } + })); + } + continue; + } + + let (org, name) = split_key(&request.key)?; + let requirement = Requirement::parse(&request.requirement); + if let Some(existing) = resolved.get(&request.key) { + if requirement.matches(&existing.version) { + continue; + } + bail!( + "version conflict for {}: resolved {} but another dependency requires `{}` (zed installs one version per package)", + request.key, + existing.version, + request.requirement + ); + } + + let version = resolve_version(registry, &org, &name, &request.requirement)?; + resolved.insert(request.key.clone(), version.clone()); + pool.submit(FetchTask { + sequence: next_sequence, + key: request.key, + version, + })?; + next_sequence += 1; + in_flight += 1; + } + + if in_flight == 0 { + break; + } + + let result = receive_in_order(pool, &mut buffered_results, next_result)?; + next_result += 1; + in_flight -= 1; + downloaded += usize::from(result.downloaded); + pending.extend( + result + .dependencies + .into_iter() + .map(|(key, requirement)| DependencyRequest { key, requirement }), + ); + } + + Ok(PrefetchReport { + resolved: resolved.len(), + downloaded, + }) + }) +} + +pub(super) fn receive_in_order( + pool: &FetchPool, + buffered: &mut BTreeMap>, + expected: usize, +) -> Result { + loop { + if let Some(result) = buffered.remove(&expected) { + return result; + } + let message = pool.receive()?; + if message.sequence == expected { + return message.result; + } + if buffered.insert(message.sequence, message.result).is_some() { + bail!( + "recursive install worker returned duplicate result sequence {}", + message.sequence + ); + } + } +} diff --git a/src/install_graph/tests.rs b/src/install_graph/tests.rs new file mode 100644 index 00000000..fbe615ca --- /dev/null +++ b/src/install_graph/tests.rs @@ -0,0 +1,711 @@ +use std::process::{Command, Stdio}; +use std::sync::{ + Arc, Barrier, + atomic::{AtomicUsize, Ordering}, + mpsc as test_mpsc, +}; +use std::thread; +use std::time::{Duration, Instant}; + +use zed_interfaces::lockfile::LockedPackage; +use zed_interfaces::registry::{PackageMetadata, VersionMetadata}; +use zed_interfaces::vcs::Vcs; + +use super::resolver::normalize_concurrency; +use super::*; +use crate::pack::pack; + +fn test_config(registry: &Path, home: &Path) -> Config { + Config { + registry: format!("file://{}", registry.display()), + home: home.to_path_buf(), + token: None, + auth_url: "http://127.0.0.1/unused".to_string(), + supabase_url: None, + supabase_key: None, + interactive: false, + } +} + +fn manifest_text(org: &str, name: &str, version: &str, dependencies: &[(&str, &str)]) -> String { + let mut text = format!( + r#"[package] +org = "{org}" +name = "{name}" +version = "{version}" + +[package.repository] +vcs = "git" +url = "https://example.invalid/{org}/{name}" +"#, + ); + if !dependencies.is_empty() { + text.push_str("\n[dependencies]\n"); + for (key, requirement) in dependencies { + text.push_str(&format!("\"{key}\" = \"{requirement}\"\n")); + } + } + text +} + +fn publish_fixture( + registry_root: &Path, + scratch: &Path, + org: &str, + name: &str, + version: &str, + dependencies: &[(&str, &str)], +) -> String { + let source = scratch.join(format!("source-{name}")); + fs::create_dir_all(&source).unwrap(); + let manifest_text = manifest_text(org, name, version, dependencies); + fs::write(source.join(MANIFEST_FILE), &manifest_text).unwrap(); + fs::write( + source.join("payload.txt"), + format!("{org}/{name}@{version}\n"), + ) + .unwrap(); + let manifest = Manifest::parse(&manifest_text).unwrap(); + let packed = pack( + &source, + &manifest, + Some(&scratch.join(format!("packed-{name}"))), + ) + .unwrap(); + + let artifacts = registry_root.join("artifacts"); + fs::create_dir_all(&artifacts).unwrap(); + fs::copy( + &packed.path, + artifacts.join(format!("{}.tar.gz", packed.sha256)), + ) + .unwrap(); + + let package_dir = registry_root.join("packages").join(org).join(name); + fs::create_dir_all(package_dir.join("versions")).unwrap(); + let version_metadata = VersionMetadata { + org: org.to_string(), + name: name.to_string(), + version: version.to_string(), + sha256: packed.sha256.clone(), + size: packed.size, + format: packed.format, + vcs_tag: format!("v{version}"), + vcs_commit: Some("0123456789abcdef0123456789abcdef01234567".to_string()), + download_url: format!("file://{}/{}.tar.gz", artifacts.display(), packed.sha256), + published_at: "1970-01-01T00:00:00Z".to_string(), + yanked: false, + }; + fs::write( + package_dir.join("versions").join(format!("{version}.json")), + serde_json::to_string_pretty(&version_metadata).unwrap(), + ) + .unwrap(); + let package_metadata = PackageMetadata { + org: org.to_string(), + name: name.to_string(), + description: Some(format!("fixture {name}")), + vcs: Vcs::Git, + repo_url: format!("https://example.invalid/{org}/{name}"), + version_scheme: manifest.package.version_scheme, + latest: Some(version.to_string()), + tags: Vec::new(), + versions: vec![version.to_string()], + }; + fs::write( + package_dir.join("package.json"), + serde_json::to_string_pretty(&package_metadata).unwrap(), + ) + .unwrap(); + packed.sha256 +} + +fn published_version( + registry_root: &Path, + org: &str, + name: &str, + version: &str, +) -> VersionMetadata { + let path = registry_root + .join("packages") + .join(org) + .join(name) + .join("versions") + .join(format!("{version}.json")); + serde_json::from_str(&fs::read_to_string(path).unwrap()).unwrap() +} + +struct CountingRegistry { + artifact: PathBuf, + downloads: Arc, + started: Option>, + release: Option>, + fail_after_write: bool, +} + +impl Registry for CountingRegistry { + fn get_package(&self, _org: &str, _name: &str) -> Result { + unreachable!("mixed-path acquisition test does not resolve package metadata") + } + + fn get_version(&self, _org: &str, _name: &str, _version: &str) -> Result { + unreachable!("mixed-path acquisition test receives resolved version metadata") + } + + fn download(&self, _version: &VersionMetadata, dest: &Path) -> Result<()> { + self.downloads.fetch_add(1, Ordering::SeqCst); + if let Some(started) = &self.started { + started.send(()).unwrap(); + } + if let Some(release) = &self.release { + release.recv_timeout(Duration::from_secs(5)).unwrap(); + } + fs::create_dir_all(dest.parent().unwrap())?; + if self.fail_after_write { + fs::write(dest, b"partial fixture download")?; + bail!("injected artifact download failure"); + } + fs::copy(&self.artifact, dest)?; + Ok(()) + } + + fn publish( + &self, + _meta: &zed_interfaces::registry::PublishMeta, + _artifact: &Path, + _token: Option<&str>, + ) -> Result { + unreachable!("mixed-path acquisition test does not publish") + } + + fn claim_org( + &self, + _slug: &str, + _token: Option<&str>, + ) -> Result { + unreachable!("mixed-path acquisition test does not claim organizations") + } + + fn search(&self, _query: &str) -> Result { + unreachable!("mixed-path acquisition test does not search") + } + + fn yank( + &self, + _org: &str, + _name: &str, + _version: &str, + _yanked: bool, + _token: Option<&str>, + ) -> Result { + unreachable!("mixed-path acquisition test does not yank") + } + + fn audit_log( + &self, + _org: &str, + _limit: Option, + _token: Option<&str>, + ) -> Result { + unreachable!("mixed-path acquisition test does not read audit logs") + } +} + +#[test] +fn legacy_and_recursive_acquisition_paths_share_one_download_lock() { + let temp = tempfile::tempdir().unwrap(); + let registry_root = temp.path().join("registry"); + let scratch = temp.path().join("scratch"); + let home = temp.path().join("home"); + + let sha = publish_fixture(®istry_root, &scratch, "test", "shared", "1.0.0", &[]); + let version = published_version(®istry_root, "test", "shared", "1.0.0"); + let artifact = registry_root + .join("artifacts") + .join(format!("{sha}.tar.gz")); + let downloads = Arc::new(AtomicUsize::new(0)); + let (recursive_started_tx, recursive_started_rx) = test_mpsc::channel(); + let (recursive_release_tx, recursive_release_rx) = test_mpsc::channel(); + let (legacy_started_tx, legacy_started_rx) = test_mpsc::channel(); + + let recursive_registry = CountingRegistry { + artifact: artifact.clone(), + downloads: Arc::clone(&downloads), + started: Some(recursive_started_tx), + release: Some(recursive_release_rx), + fail_after_write: false, + }; + let legacy_registry = CountingRegistry { + artifact, + downloads: Arc::clone(&downloads), + started: Some(legacy_started_tx), + release: None, + fail_after_write: false, + }; + let recursive_store = Store::new(&home); + let legacy_store = Store::new(&home); + let recursive_version = version.clone(); + let legacy_version = version; + + let recursive = thread::spawn(move || { + super::ensure_artifact(&recursive_registry, &recursive_store, &recursive_version).unwrap() + }); + recursive_started_rx + .recv_timeout(Duration::from_secs(2)) + .unwrap(); + + let legacy = thread::spawn(move || { + crate::ops::legacy_ensure_artifact_for_test( + &legacy_registry, + &legacy_store, + &legacy_version, + ) + .unwrap() + }); + + let legacy_entered_download = legacy_started_rx + .recv_timeout(Duration::from_millis(300)) + .is_ok(); + recursive_release_tx.send(()).unwrap(); + + let (recursive_path, recursive_downloaded) = recursive.join().unwrap(); + let legacy_path = legacy.join().unwrap(); + assert!(recursive_downloaded); + assert!( + !legacy_entered_download, + "the legacy path bypassed the shared artifact lock and started a duplicate download" + ); + assert_eq!(downloads.load(Ordering::SeqCst), 1); + assert_eq!(legacy_path, recursive_path); + assert!(recursive_path.is_dir()); +} + +#[test] +fn worker_panics_are_reported_without_losing_the_task_sequence() { + let message = super::artifact::run_fetch_task(17, "test/panic", || -> Result { + panic!("fixture worker panic") + }); + + assert_eq!(message.sequence, 17); + let error = message.result.unwrap_err().to_string(); + assert!(error.contains("test/panic"), "{error}"); + assert!(error.contains("sequence 17"), "{error}"); + assert!(error.contains("fixture worker panic"), "{error}"); +} + +#[test] +fn failed_downloads_remove_staging_files_and_never_publish_cache_entries() { + let temp = tempfile::tempdir().unwrap(); + let registry_root = temp.path().join("registry"); + let scratch = temp.path().join("scratch"); + let home = temp.path().join("home"); + + let sha = publish_fixture(®istry_root, &scratch, "test", "failure", "1.0.0", &[]); + let version = published_version(®istry_root, "test", "failure", "1.0.0"); + let registry = CountingRegistry { + artifact: registry_root + .join("artifacts") + .join(format!("{sha}.tar.gz")), + downloads: Arc::new(AtomicUsize::new(0)), + started: None, + release: None, + fail_after_write: true, + }; + let store = Store::new(&home); + + let error = super::ensure_artifact(®istry, &store, &version) + .unwrap_err() + .to_string(); + assert!( + error.contains("injected artifact download failure"), + "{error}" + ); + assert!(!store.cached_artifact(&sha).exists()); + assert!(!store.has(&sha)); + let cache_entries = fs::read_dir(store.cache_dir()) + .map(|entries| entries.count()) + .unwrap_or(0); + assert_eq!(cache_entries, 0, "failed staging directory leaked"); +} + +#[test] +fn defaults_to_five_workers_and_bounds_overrides() { + assert_eq!(normalize_concurrency(None), 5); + assert_eq!(normalize_concurrency(Some("not-a-number")), 5); + assert_eq!(normalize_concurrency(Some("0")), 5); + assert_eq!(normalize_concurrency(Some("1")), 1); + assert_eq!(normalize_concurrency(Some("999")), MAX_INSTALL_CONCURRENCY); +} + +#[test] +fn recursively_prefetches_packages_of_packages_and_deduplicates_diamonds() { + let temp = tempfile::tempdir().unwrap(); + let registry = temp.path().join("registry"); + let scratch = temp.path().join("scratch"); + let home = temp.path().join("home"); + let project = temp.path().join("project"); + fs::create_dir_all(&project).unwrap(); + let leaf = publish_fixture(®istry, &scratch, "test", "leaf", "1.0.0", &[]); + let left = publish_fixture( + ®istry, + &scratch, + "test", + "left", + "1.0.0", + &[("test/leaf", "^1")], + ); + let right = publish_fixture( + ®istry, + &scratch, + "test", + "right", + "1.0.0", + &[("test/leaf", "^1")], + ); + let root = publish_fixture( + ®istry, + &scratch, + "test", + "root", + "1.0.0", + &[("test/left", "^1"), ("test/right", "^1")], + ); + fs::write( + project.join(MANIFEST_FILE), + manifest_text("consumer", "app", "0.1.0", &[("test/root", "^1")]), + ) + .unwrap(); + + let report = prefetch(&project, &test_config(®istry, &home), false).unwrap(); + assert_eq!(report.resolved, 4); + assert_eq!(report.downloaded, 4); + let store = Store::new(&home); + for sha in [leaf, left, right, root] { + assert!(store.has(&sha), "missing prefetched artifact {sha}"); + } +} + +#[test] +fn dependency_cycles_terminate_after_each_package_is_resolved_once() { + let temp = tempfile::tempdir().unwrap(); + let registry = temp.path().join("registry"); + let scratch = temp.path().join("scratch"); + let home = temp.path().join("home"); + let project = temp.path().join("project"); + fs::create_dir_all(&project).unwrap(); + + publish_fixture( + ®istry, + &scratch, + "test", + "a", + "1.0.0", + &[("test/b", "^1")], + ); + publish_fixture( + ®istry, + &scratch, + "test", + "b", + "1.0.0", + &[("test/a", "^1")], + ); + fs::write( + project.join(MANIFEST_FILE), + manifest_text("consumer", "app", "0.1.0", &[("test/a", "^1")]), + ) + .unwrap(); + + let report = prefetch(&project, &test_config(®istry, &home), false).unwrap(); + assert_eq!(report.resolved, 2); + assert_eq!(report.downloaded, 2); +} + +#[test] +fn concurrent_prefetches_share_one_artifact_download() { + let temp = tempfile::tempdir().unwrap(); + let registry = temp.path().join("registry"); + let scratch = temp.path().join("scratch"); + let home = temp.path().join("home"); + let first_project = temp.path().join("first-project"); + let second_project = temp.path().join("second-project"); + fs::create_dir_all(&first_project).unwrap(); + fs::create_dir_all(&second_project).unwrap(); + + publish_fixture(®istry, &scratch, "test", "leaf", "1.0.0", &[]); + let consumer_manifest = manifest_text("consumer", "app", "0.1.0", &[("test/leaf", "^1")]); + fs::write(first_project.join(MANIFEST_FILE), &consumer_manifest).unwrap(); + fs::write(second_project.join(MANIFEST_FILE), &consumer_manifest).unwrap(); + + let barrier = Arc::new(Barrier::new(3)); + let first_cfg = test_config(®istry, &home); + let second_cfg = first_cfg.clone(); + let first_barrier = Arc::clone(&barrier); + let second_barrier = Arc::clone(&barrier); + let first = thread::spawn(move || { + first_barrier.wait(); + prefetch(&first_project, &first_cfg, false).unwrap() + }); + let second = thread::spawn(move || { + second_barrier.wait(); + prefetch(&second_project, &second_cfg, false).unwrap() + }); + + barrier.wait(); + let first = first.join().unwrap(); + let second = second.join().unwrap(); + assert_eq!(first.resolved + second.resolved, 2); + assert_eq!( + first.downloaded + second.downloaded, + 1, + "one process should publish the shared content hash and the other should reuse it" + ); +} + +#[test] +fn a_warm_store_resolves_the_graph_without_redownloading_artifacts() { + let temp = tempfile::tempdir().unwrap(); + let registry = temp.path().join("registry"); + let scratch = temp.path().join("scratch"); + let home = temp.path().join("home"); + let project = temp.path().join("project"); + fs::create_dir_all(&project).unwrap(); + + publish_fixture(®istry, &scratch, "test", "leaf", "1.0.0", &[]); + publish_fixture( + ®istry, + &scratch, + "test", + "left", + "1.0.0", + &[("test/leaf", "^1")], + ); + publish_fixture( + ®istry, + &scratch, + "test", + "right", + "1.0.0", + &[("test/leaf", "^1")], + ); + publish_fixture( + ®istry, + &scratch, + "test", + "root", + "1.0.0", + &[("test/left", "^1"), ("test/right", "^1")], + ); + fs::write( + project.join(MANIFEST_FILE), + manifest_text("consumer", "app", "0.1.0", &[("test/root", "^1")]), + ) + .unwrap(); + let cfg = test_config(®istry, &home); + + let cold = prefetch(&project, &cfg, false).unwrap(); + let warm = prefetch(&project, &cfg, false).unwrap(); + assert_eq!(cold.resolved, 4); + assert_eq!(cold.downloaded, 4); + assert_eq!(warm.resolved, 4); + assert_eq!(warm.downloaded, 0); +} + +#[test] +fn conflicting_transitive_requirements_fail_in_deterministic_graph_order() { + let temp = tempfile::tempdir().unwrap(); + let registry = temp.path().join("registry"); + let scratch = temp.path().join("scratch"); + let home = temp.path().join("home"); + let project = temp.path().join("project"); + fs::create_dir_all(&project).unwrap(); + + publish_fixture(®istry, &scratch, "test", "leaf", "1.0.0", &[]); + publish_fixture( + ®istry, + &scratch, + "test", + "left", + "1.0.0", + &[("test/leaf", "^1")], + ); + publish_fixture( + ®istry, + &scratch, + "test", + "right", + "1.0.0", + &[("test/leaf", "^2")], + ); + publish_fixture( + ®istry, + &scratch, + "test", + "root", + "1.0.0", + &[("test/left", "^1"), ("test/right", "^1")], + ); + fs::write( + project.join(MANIFEST_FILE), + manifest_text("consumer", "app", "0.1.0", &[("test/root", "^1")]), + ) + .unwrap(); + + let error = prefetch(&project, &test_config(®istry, &home), false).unwrap_err(); + let message = format!("{error:#}"); + assert!( + message.contains("version conflict for test/leaf"), + "{message}" + ); + assert!(message.contains("resolved 1.0.0"), "{message}"); + assert!(message.contains("requires `^2`"), "{message}"); +} + +#[test] +fn frozen_prefetch_installs_every_locked_artifact_without_a_manifest() { + let temp = tempfile::tempdir().unwrap(); + let registry = temp.path().join("registry"); + let scratch = temp.path().join("scratch"); + let home = temp.path().join("home"); + let project = temp.path().join("project"); + fs::create_dir_all(&project).unwrap(); + + for (name, dependencies) in [ + ("leaf", vec![]), + ("left", vec![("test/leaf", "^1")]), + ("right", vec![("test/leaf", "^1")]), + ("root", vec![("test/left", "^1"), ("test/right", "^1")]), + ] { + publish_fixture(®istry, &scratch, "test", name, "1.0.0", &dependencies); + } + + let source = format!("file://{}", registry.display()); + let mut lock = Lockfile::default(); + for name in ["leaf", "left", "right", "root"] { + let version = published_version(®istry, "test", name, "1.0.0"); + lock.upsert(LockedPackage { + org: version.org, + name: version.name, + version: version.version, + sha256: version.sha256, + size: version.size, + format: version.format, + vcs_tag: version.vcs_tag, + vcs_commit: version.vcs_commit, + source: source.clone(), + }); + } + fs::write(project.join(LOCKFILE_FILE), lock.to_toml_string().unwrap()).unwrap(); + + let report = prefetch(&project, &test_config(®istry, &home), true).unwrap(); + assert_eq!(report.resolved, 4); + assert_eq!(report.downloaded, 4); +} + +#[test] +fn a_corrupt_partial_cache_is_replaced_under_the_artifact_lock() { + let temp = tempfile::tempdir().unwrap(); + let registry = temp.path().join("registry"); + let scratch = temp.path().join("scratch"); + let home = temp.path().join("home"); + let project = temp.path().join("project"); + fs::create_dir_all(&project).unwrap(); + + let sha = publish_fixture(®istry, &scratch, "test", "leaf", "1.0.0", &[]); + fs::write( + project.join(MANIFEST_FILE), + manifest_text("consumer", "app", "0.1.0", &[("test/leaf", "^1")]), + ) + .unwrap(); + let store = Store::new(&home); + let cached = store.cached_artifact(&sha); + fs::create_dir_all(cached.parent().unwrap()).unwrap(); + fs::write(&cached, b"partial download").unwrap(); + + let report = prefetch(&project, &test_config(®istry, &home), false).unwrap(); + assert_eq!(report.resolved, 1); + assert_eq!(report.downloaded, 1); + assert!(store.has(&sha)); + assert_eq!(sha256_file(&cached).unwrap().0, sha); +} + +#[test] +fn artifact_waiters_block_until_the_owner_releases_the_kernel_lock() { + const CHILD_HOME: &str = "ZED_PKG_TEST_LOCK_CHILD_HOME"; + const CHILD_SHA: &str = "ZED_PKG_TEST_LOCK_CHILD_SHA"; + const CHILD_ATTEMPTING: &str = "ZED_PKG_TEST_LOCK_CHILD_ATTEMPTING"; + const CHILD_ACQUIRED: &str = "ZED_PKG_TEST_LOCK_CHILD_ACQUIRED"; + const TEST_NAME: &str = + "install_graph::tests::artifact_waiters_block_until_the_owner_releases_the_kernel_lock"; + + if let Some(home) = std::env::var_os(CHILD_HOME) { + let sha = std::env::var(CHILD_SHA).unwrap(); + let attempting = PathBuf::from(std::env::var_os(CHILD_ATTEMPTING).unwrap()); + let acquired = PathBuf::from(std::env::var_os(CHILD_ACQUIRED).unwrap()); + fs::write(attempting, b"attempting").unwrap(); + let _waiter = ArtifactProcessLock::acquire(Path::new(&home), &sha).unwrap(); + fs::write(acquired, b"acquired").unwrap(); + return; + } + + let temp = tempfile::tempdir().unwrap(); + let sha = "a".repeat(64); + let owner = ArtifactProcessLock::acquire(temp.path(), &sha).unwrap(); + let attempting = temp.path().join("waiter-attempting"); + let acquired = temp.path().join("waiter-acquired"); + let mut child = Command::new(std::env::current_exe().unwrap()) + .arg(TEST_NAME) + .arg("--exact") + .arg("--nocapture") + .env(CHILD_HOME, temp.path()) + .env(CHILD_SHA, &sha) + .env(CHILD_ATTEMPTING, &attempting) + .env(CHILD_ACQUIRED, &acquired) + .stdin(Stdio::null()) + .stdout(Stdio::inherit()) + .stderr(Stdio::inherit()) + .spawn() + .unwrap(); + + let attempting_deadline = Instant::now() + Duration::from_secs(5); + while !attempting.is_file() { + if let Some(status) = child.try_wait().unwrap() { + panic!("lock waiter child exited before attempting acquisition: {status}"); + } + if Instant::now() >= attempting_deadline { + let _ = child.kill(); + let _ = child.wait(); + panic!("lock waiter child did not reach acquisition"); + } + thread::sleep(Duration::from_millis(20)); + } + + thread::sleep(Duration::from_millis(200)); + assert!( + !acquired.exists(), + "a separate process acquired the artifact lock while the owner still held it" + ); + assert!( + child.try_wait().unwrap().is_none(), + "lock waiter child exited before the owner released the lock" + ); + + drop(owner); + let release_deadline = Instant::now() + Duration::from_secs(5); + loop { + if let Some(status) = child.try_wait().unwrap() { + assert!(status.success(), "lock waiter child failed: {status}"); + break; + } + if Instant::now() >= release_deadline { + let _ = child.kill(); + let _ = child.wait(); + panic!("lock waiter child did not wake after lock release"); + } + thread::sleep(Duration::from_millis(20)); + } + assert!( + acquired.is_file(), + "lock waiter child exited without recording acquisition" + ); +} diff --git a/src/lib.rs b/src/lib.rs index f424fc94..488060ef 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -8,10 +8,12 @@ pub mod config; pub mod dev; pub mod fetch; pub mod flags; +pub mod install_graph; pub mod interactive; pub mod manifestless; pub mod nix_export_bundle; pub mod nix_export_plan; +#[path = "ops_entry.rs"] pub mod ops; pub mod pack; pub mod preflight; diff --git a/src/ops.rs b/src/ops.rs index eed9e201..bd4b1cf9 100644 --- a/src/ops.rs +++ b/src/ops.rs @@ -219,6 +219,51 @@ fn collect_members(root: &Path, globs: &[String]) -> WorkspaceInfo { info } +fn collect_workspace_links_for_frozen( + project: &Path, + manifest: &Manifest, + workspace: Option<&WorkspaceInfo>, +) -> Result> { + let Some(workspace) = workspace else { + return Ok(BTreeMap::new()); + }; + + let mut links = BTreeMap::new(); + let mut pending: VecDeque<(String, String)> = manifest + .dependencies + .iter() + .map(|(key, requirement)| (key.clone(), requirement.clone())) + .collect(); + + while let Some((raw_key, requirement_text)) = pending.pop_front() { + let (org, name) = split_key(&raw_key)?; + let key = format!("{org}/{name}"); + let Some(member_dir) = workspace.members.get(&key) else { + continue; + }; + let member_manifest = read_manifest(member_dir).with_context(|| { + format!( + "reading workspace member `{key}` from {}", + member_dir.display() + ) + })?; + let requirement = Requirement::parse(&requirement_text); + if !requirement.matches(&member_manifest.package.version) { + bail!( + "workspace member {key}@{} does not satisfy `{requirement_text}`", + member_manifest.package.version + ); + } + if member_dir == project || links.contains_key(&key) { + continue; + } + links.insert(key, member_dir.clone()); + pending.extend(member_manifest.dependencies); + } + + Ok(links) +} + // --------------------------------------------------------------------------- // install @@ -229,14 +274,17 @@ pub struct InstallOutcome { fn ensure_artifact(reg: &dyn Registry, store: &Store, vm: &VersionMetadata) -> Result { validate_version_metadata(vm)?; - if store.has(&vm.sha256) { - return Ok(store.pkg_dir(&vm.sha256)); - } - let cached = store.cached_artifact(&vm.sha256); - if !cached.exists() { - reg.download(vm, &cached)?; - } - store.add_artifact(&cached, &vm.sha256) + crate::install_graph::ensure_artifact(reg, store, vm) + .map(|(package_dir, _downloaded)| package_dir) +} + +#[cfg(test)] +pub(crate) fn legacy_ensure_artifact_for_test( + reg: &dyn Registry, + store: &Store, + vm: &VersionMetadata, +) -> Result { + ensure_artifact(reg, store, vm) } fn replace_dest(dest: &Path) -> Result<()> { @@ -1071,6 +1119,8 @@ fn install_locked( workspace.as_ref(), validate_manifest_requirements, )?; + workspace_links = + collect_workspace_links_for_frozen(project, &manifest, workspace.as_ref())?; for locked in &lock.packages { if !is_slug(&locked.org) || !is_slug(&locked.name) { bail!( @@ -1106,13 +1156,24 @@ fn install_locked( if let Some(ws) = &workspace && let Some(member_dir) = ws.members.get(&key) { + let member_manifest = read_manifest(member_dir).with_context(|| { + format!( + "reading workspace member `{key}` from {}", + member_dir.display() + ) + })?; + let requirement = Requirement::parse(&req_str); + if !requirement.matches(&member_manifest.package.version) { + bail!( + "workspace member {key}@{} does not satisfy `{req_str}`", + member_manifest.package.version + ); + } if member_dir != project && !workspace_links.contains_key(&key) { workspace_links.insert(key.clone(), member_dir.clone()); - if let Ok(member_manifest) = read_manifest(member_dir) { - for (sub_key, sub_req) in member_manifest.dependencies { - let (sub_org, sub_name) = split_key(&sub_key)?; - queue.push_back((sub_org, sub_name, sub_req)); - } + for (sub_key, sub_req) in member_manifest.dependencies { + let (sub_org, sub_name) = split_key(&sub_key)?; + queue.push_back((sub_org, sub_name, sub_req)); } } continue; @@ -1364,16 +1425,95 @@ fn install_locked( for (key, member_dir) in &workspace_links { interactive::confirm(cfg.interactive, &format!("link workspace package {key}"))?; let (org, name) = split_key(key)?; + let member_manifest = read_manifest(member_dir).with_context(|| { + format!( + "reading workspace member `{key}` from {}", + member_dir.display() + ) + })?; + if !allow_ecosystem_mismatch + && let Some(problem) = ecosystem_mismatch( + key, + &name, + member_manifest.package.language, + member_manifest.package.ecosystem(), + &project_ecos, + ) + { + bail!("{problem}"); + } + let dest = modules.join(&org).join(&name); // Workspace dependencies obey the same ownership decision as registry // packages. Copy mode must remain self-contained after the member source // directory leaves the Docker/OCI build context. link_or_copy(member_dir, &dest, mode)?; - if let Ok(member_manifest) = read_manifest(member_dir) { - for (bin_name, rel_target) in &member_manifest.bin { - bins.insert(bin_name.clone(), dest.join(rel_target)); + for (bin_name, rel_target) in &member_manifest.bin { + bins.insert(bin_name.clone(), dest.join(rel_target)); + } + + let dependency_adapter = if use_dependency_adapters { + member_manifest + .install + .adapter + .as_deref() + .map(named_adapter) + .transpose()? + .unwrap_or(adapter) + } else { + adapter + }; + match dependency_adapter { + Adapter::Node => { + used_node_adapter = true; + let node_dest = project + .join("node_modules") + .join(format!("@{org}")) + .join(&name); + transaction.backup(&node_dest)?; + link_or_copy(member_dir, &node_dest, mode)?; + } + Adapter::Java => { + used_java_adapter = true; + for entry in walkdir::WalkDir::new(&dest) + .follow_links(true) + .into_iter() + .filter_map(|entry| entry.ok()) + { + if entry + .path() + .extension() + .is_some_and(|extension| extension == "jar") + { + jars.push(entry.path().to_string_lossy().to_string()); + } + } } + Adapter::Go => wired_roots + .entry(Adapter::Go) + .or_default() + .push(dest.clone()), + Adapter::Python => wired_roots + .entry(Adapter::Python) + .or_default() + .push(dest.clone()), + Adapter::Rust => wired_roots + .entry(Adapter::Rust) + .or_default() + .push(dest.clone()), + Adapter::Dart => wired_roots + .entry(Adapter::Dart) + .or_default() + .push(dest.clone()), + Adapter::Auto | Adapter::None => {} } + wired_packages.push(WiredPackage { + key: key.clone(), + version: "workspace".to_string(), + language: member_manifest.package.language, + ecosystem: member_manifest.package.ecosystem(), + path: dest, + }); installed.push((key.clone(), "workspace".to_string())); } hoist_bins(&modules, &bins, mode)?; @@ -1457,7 +1597,22 @@ pub fn uninstall(project: &Path, cfg: &Config, specs: &[String]) -> Result<()> { let text = fs::read_to_string(&lock_path) .with_context(|| format!("zed uninstall requires {LOCKFILE_FILE}"))?; let lock = Lockfile::parse(&text).with_context(|| format!("invalid {LOCKFILE_FILE}"))?; - if lock.packages.is_empty() { + + // Workspace packages deliberately do not appear in the artifact lock: the + // lock records immutable registry hashes, while workspace members are live + // source projections. Reconstruct the active workspace graph from the same + // manifest boundary used by frozen install so a workspace-only project can + // still uninstall and later restore its exact materialized graph. + let manifest = read_manifest(project).ok(); + let workspace = find_workspace(project); + let workspace_links = match manifest.as_ref() { + Some(manifest) => { + collect_workspace_links_for_frozen(project, manifest, workspace.as_ref())? + } + None => BTreeMap::new(), + }; + let total_materialized = lock.packages.len() + workspace_links.len(); + if total_materialized == 0 { println!("nothing to uninstall"); return Ok(()); } @@ -1465,6 +1620,7 @@ pub fn uninstall(project: &Path, cfg: &Config, specs: &[String]) -> Result<()> { let mut targets = BTreeSet::new(); if specs.is_empty() { targets.extend(lock.packages.iter().map(LockedPackage::full_name)); + targets.extend(workspace_links.keys().cloned()); } else { for spec in specs { if spec.contains('@') { @@ -1472,8 +1628,13 @@ pub fn uninstall(project: &Path, cfg: &Config, specs: &[String]) -> Result<()> { } let (org, name) = split_key(spec)?; let key = format!("{org}/{name}"); + if workspace_links.contains_key(&key) { + bail!( + "selective uninstall of workspace package `{key}` is not supported; run `zed uninstall` without package arguments to remove the complete materialized graph while retaining {LOCKFILE_FILE}" + ); + } if lock.find(&org, &name).is_none() { - bail!("{key} is not pinned by {LOCKFILE_FILE}"); + bail!("{key} is neither pinned by {LOCKFILE_FILE} nor an active workspace package"); } targets.insert(key); } @@ -1490,7 +1651,6 @@ pub fn uninstall(project: &Path, cfg: &Config, specs: &[String]) -> Result<()> { let store = Store::new(&cfg.home); let _install_lock = store.install_lock()?; - let manifest = read_manifest(project).ok(); let modules_dir = manifest .as_ref() .map(|manifest| manifest.modules_dir().to_string()) @@ -1498,7 +1658,7 @@ pub fn uninstall(project: &Path, cfg: &Config, specs: &[String]) -> Result<()> { let modules = project.join(&modules_dir); let had_node_adapter = project.join(".zed").join("node_path").is_file(); let had_java_adapter = project.join(".zed").join("classpath").is_file(); - let uninstall_all = targets.len() == lock.packages.len(); + let uninstall_all = targets.len() == total_materialized; let mut transaction = ProjectTransaction::begin(project)?; eprintln!( @@ -1507,14 +1667,25 @@ pub fn uninstall(project: &Path, cfg: &Config, specs: &[String]) -> Result<()> { ); if uninstall_all { transaction.backup(&modules)?; + // These files are all generated projections of the materialized graph. + // Remove them transactionally on a full uninstall so no toolchain sees + // stale package paths while the retained lock waits for frozen restore. + for generated in [ + "paths.json", + "node_path", + "classpath", + "go.work", + "pythonpath", + "cargo-paths.toml", + "pub-deps.yaml", + ] { + transaction.backup(&project.join(".zed").join(generated))?; + } } else { transaction.backup(&modules.join(BIN_DIR))?; - } - if had_java_adapter { - transaction.backup(&project.join(".zed").join("classpath"))?; - } - if uninstall_all && had_node_adapter { - transaction.backup(&project.join(".zed").join("node_path"))?; + if had_java_adapter { + transaction.backup(&project.join(".zed").join("classpath"))?; + } } for key in &targets { @@ -1538,6 +1709,10 @@ pub fn uninstall(project: &Path, cfg: &Config, specs: &[String]) -> Result<()> { .iter() .filter(|package| !targets.contains(&package.full_name())) .collect(); + let remaining_workspace = workspace_links + .keys() + .filter(|key| !targets.contains(*key)) + .count(); if !uninstall_all { let mut bins: BTreeMap = BTreeMap::new(); let mut mode = InstallMode::Copy; @@ -1570,20 +1745,52 @@ pub fn uninstall(project: &Path, cfg: &Config, specs: &[String]) -> Result<()> { } } } + // Selective uninstall currently targets registry packages only. Keep + // workspace bins and Java entries in the rebuilt aggregate projections. + for key in workspace_links.keys().filter(|key| !targets.contains(*key)) { + let (org, name) = split_key(key)?; + let installed = modules.join(&org).join(&name); + if fs::symlink_metadata(&installed) + .is_ok_and(|metadata| metadata.file_type().is_symlink()) + { + mode = InstallMode::Symlink; + } + if let Ok(package_manifest) = read_manifest(&installed) { + for (bin_name, target) in package_manifest.bin { + bins.insert(bin_name, installed.join(target)); + } + } + if had_java_adapter && installed.exists() { + for entry in walkdir::WalkDir::new(&installed) + .follow_links(true) + .into_iter() + .filter_map(|entry| entry.ok()) + { + if entry + .path() + .extension() + .is_some_and(|extension| extension == "jar") + { + jars.push(entry.path().to_string_lossy().to_string()); + } + } + } + } hoist_bins(&modules, &bins, mode)?; if had_java_adapter && !jars.is_empty() { jars.sort(); + jars.dedup(); let classpath = project.join(".zed").join("classpath"); fs::create_dir_all(classpath.parent().context("classpath parent")?)?; fs::write(classpath, jars.join(":") + "\n")?; } } + let remaining_total = remaining.len() + remaining_workspace; interactive::confirm( cfg.interactive, &format!( - "record {} remaining installed package(s) and commit transaction {}", - remaining.len(), + "record {remaining_total} remaining installed package(s) and commit transaction {}", transaction.id() ), )?; @@ -1600,8 +1807,7 @@ pub fn uninstall(project: &Path, cfg: &Config, specs: &[String]) -> Result<()> { println!("uninstalled {key}"); } println!( - "{} package(s) remain materialized; {LOCKFILE_FILE} retained for frozen reinstall", - remaining.len() + "{remaining_total} package(s) remain materialized; {LOCKFILE_FILE} retained for frozen reinstall" ); Ok(()) } @@ -2700,6 +2906,105 @@ version = "1.0.0" assert!(error.contains("two paths"), "{error}"); } + #[test] + fn frozen_workspace_resolution_expands_transitive_members_and_validates_versions() { + let temp = tempfile::tempdir().unwrap(); + let root = temp.path(); + let app = root.join("apps/cli"); + let utils = root.join("packages/utils"); + let core = root.join("packages/core"); + for directory in [&app, &utils, &core] { + fs::create_dir_all(directory).unwrap(); + } + + fs::write( + root.join(MANIFEST_FILE), + r#"[package] +org = "zedtest" +name = "workspace-root" +version = "1.0.0" + +[package.repository] +vcs = "git" +url = "https://example.invalid/workspace-root" + +[workspace] +members = ["packages/*", "apps/*"] +"#, + ) + .unwrap(); + fs::write( + core.join(MANIFEST_FILE), + r#"[package] +org = "zedtest" +name = "ws-core" +version = "1.2.0" + +[package.repository] +vcs = "git" +url = "https://example.invalid/ws-core" +"#, + ) + .unwrap(); + fs::write( + utils.join(MANIFEST_FILE), + r#"[package] +org = "zedtest" +name = "ws-utils" +version = "1.1.0" + +[package.repository] +vcs = "git" +url = "https://example.invalid/ws-utils" + +[dependencies] +"zedtest/ws-core" = "^1" +"#, + ) + .unwrap(); + fs::write( + app.join(MANIFEST_FILE), + r#"[package] +org = "zedtest" +name = "ws-cli" +version = "1.0.0" + +[package.repository] +vcs = "git" +url = "https://example.invalid/ws-cli" + +[dependencies] +"zedtest/ws-utils" = "^1" +"#, + ) + .unwrap(); + + let manifest = read_manifest(&app).unwrap(); + let workspace = find_workspace(&app).unwrap(); + let links = collect_workspace_links_for_frozen(&app, &manifest, Some(&workspace)).unwrap(); + assert_eq!( + links.keys().cloned().collect::>(), + vec![ + "zedtest/ws-core".to_string(), + "zedtest/ws-utils".to_string() + ] + ); + assert_eq!(links["zedtest/ws-core"], core); + assert_eq!(links["zedtest/ws-utils"], utils); + + let incompatible = manifest + .to_toml_string() + .unwrap() + .replace("\"^1\"", "\"^2\""); + fs::write(app.join(MANIFEST_FILE), incompatible).unwrap(); + let manifest = read_manifest(&app).unwrap(); + let error = collect_workspace_links_for_frozen(&app, &manifest, Some(&workspace)) + .unwrap_err() + .to_string(); + assert!(error.contains("ws-utils@1.1.0"), "{error}"); + assert!(error.contains("does not satisfy `^2`"), "{error}"); + } + #[test] fn lock_only_frozen_restore_skips_only_the_missing_manifest_comparison() { let manifest = Manifest::parse( diff --git a/src/ops_entry.rs b/src/ops_entry.rs new file mode 100644 index 00000000..36815c27 --- /dev/null +++ b/src/ops_entry.rs @@ -0,0 +1,83 @@ +//! Public installer facade. +//! +//! The existing implementation continues to own dependency resolution, +//! project transactions, lockfile writes, adapter wiring, and materialization. +//! This facade warms the content-addressed store with the recursive, +//! bounded-concurrency installer before entering that transactional phase. + +use std::path::Path; + +use anyhow::Result; + +use crate::cli::{Adapter, InstallMode}; +use crate::config::Config; + +#[path = "ops.rs"] +mod implementation; + +pub use implementation::{ + InstallOutcome, WorkspaceInfo, build_cmd, build_publish_meta, cache_clean, find, gc, init, + login, org_audit, org_claim, pack_cmd, publish, run, split_key, store_prune, store_status, + uninstall, yank, +}; + +pub(crate) use implementation::{ + detect_adapter, detect_native_manifest_target, detect_structure_target, detect_target, +}; + +#[cfg(test)] +pub(crate) use implementation::legacy_ensure_artifact_for_test; + +pub fn add(project: &Path, cfg: &Config, spec: &str) -> Result<()> { + crate::config::with_install_prefetch(cfg, || implementation::add(project, cfg, spec)) +} + +pub fn remove(project: &Path, cfg: &Config, spec: &str) -> Result<()> { + crate::config::with_install_prefetch(cfg, || implementation::remove(project, cfg, spec)) +} + +#[allow(clippy::too_many_arguments)] +pub fn install( + project: &Path, + cfg: &Config, + frozen: bool, + mode: InstallMode, + adapter: Adapter, + allow_build: bool, + target: Option<&str>, + allow_ecosystem_mismatch: bool, +) -> Result { + crate::install_graph::prefetch(project, cfg, frozen)?; + implementation::install( + project, + cfg, + frozen, + mode, + adapter, + allow_build, + target, + allow_ecosystem_mismatch, + ) +} + +#[allow(clippy::too_many_arguments)] +pub(crate) fn install_frozen_lock_only( + project: &Path, + cfg: &Config, + mode: InstallMode, + adapter: Adapter, + allow_build: bool, + target: Option<&str>, + allow_ecosystem_mismatch: bool, +) -> Result { + crate::install_graph::prefetch(project, cfg, true)?; + implementation::install_frozen_lock_only( + project, + cfg, + mode, + adapter, + allow_build, + target, + allow_ecosystem_mismatch, + ) +} diff --git a/src/store.rs b/src/store.rs index 6f4dfcce..60c03286 100644 --- a/src/store.rs +++ b/src/store.rs @@ -2,7 +2,7 @@ use std::collections::{BTreeMap, BTreeSet}; use std::fs; use std::io::Read; use std::path::{Component, Path, PathBuf}; -use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH}; +use std::time::{Duration, SystemTime, UNIX_EPOCH}; use anyhow::{Context, Result, bail}; use flate2::read::GzDecoder; @@ -55,22 +55,18 @@ fn require_build_key(key: &str) -> Result<()> { Ok(()) } -/// An advisory (flock-based) process lock held for the life of the guard. -/// CLI commands are highly concurrent — two `zed install` runs in different -/// terminals, or N parallel CI runners — so store mutations that must not -/// interleave (extracting the same artifact, rewriting refs.json) take a -/// lock first. The OS releases it if the process dies, so a crash can't -/// wedge the store. +/// A descriptor-backed process lock held for the life of the guard. +/// +/// Acquisition uses the operating system's blocking lock primitive directly: +/// `flock`/`fcntl` semantics on Unix and `LockFileEx` semantics on Windows via +/// `fs2`. Contended callers sleep in the kernel and wake when the owner drops +/// the descriptor or exits. There is no retry timer, jitter loop, stale-file +/// reclamation, or userspace polling. pub struct ProcessLock { _file: fs::File, } impl ProcessLock { - /// Three-phase acquisition tuned for CLI latency: - /// spin-wait at 50ms while young (catches sub-second releases with no - /// perceptible lag), then exponential backoff with full jitter so - /// parallel CI runners don't wake in lockstep, and after 3s tell the - /// user what they are waiting on instead of freezing silently. fn acquire(path: &Path, waiting_on: &str) -> Result { if let Some(parent) = path.parent() { fs::create_dir_all(parent)?; @@ -83,52 +79,13 @@ impl ProcessLock { .open(path) .with_context(|| format!("opening lock file {}", path.display()))?; - let timeout = Duration::from_secs( - std::env::var("ZED_PKG_LOCK_TIMEOUT") - .ok() - .and_then(|v| v.parse().ok()) - .unwrap_or(600), - ); - let start = Instant::now(); - let mut backoff_attempt: u32 = 0; - let mut notified = false; - loop { - match file.try_lock_exclusive() { - Ok(()) => return Ok(Self { _file: file }), - Err(_) if start.elapsed() >= timeout => { - bail!( - "timed out after {}s waiting for {waiting_on} (lock {}); \ - another zed process may be stuck — raise ZED_PKG_LOCK_TIMEOUT \ - or investigate", - timeout.as_secs(), - path.display() - ); - } - Err(_) => {} - } - let elapsed = start.elapsed(); - if elapsed >= Duration::from_secs(3) && !notified { - eprintln!("waiting for {waiting_on} (held by another zed process)..."); - notified = true; - } - let sleep = if elapsed < Duration::from_millis(500) { - Duration::from_millis(50) - } else { - backoff_attempt = backoff_attempt.saturating_add(1); - let cap_ms: u64 = 2000; - let exp_ms = 100u64 - .saturating_mul(1u64 << backoff_attempt.min(20)) - .min(cap_ms); - // Full jitter: uniform in [0, exp_ms] without pulling in an - // RNG dependency — subsec_nanos is plenty for de-syncing. - let nanos = SystemTime::now() - .duration_since(UNIX_EPOCH) - .map(|d| d.subsec_nanos() as u64) - .unwrap_or(0); - Duration::from_millis(nanos % (exp_ms + 1)) - }; - std::thread::sleep(sleep); - } + file.lock_exclusive().with_context(|| { + format!( + "waiting for {waiting_on} through operating-system lock {}", + path.display() + ) + })?; + Ok(Self { _file: file }) } } @@ -161,6 +118,10 @@ impl Store { } } + pub(crate) fn home(&self) -> &Path { + &self.home + } + pub fn root(&self) -> PathBuf { self.home.join("store") } @@ -684,8 +645,222 @@ pub fn human_size(bytes: u64) -> String { #[cfg(test)] mod tests { + use std::process::{Command, Stdio}; + use std::time::Instant; + use super::*; + #[test] + fn install_process_lock_blocks_without_polling_until_owner_release() { + const CHILD_HOME: &str = "ZED_PKG_TEST_STORE_LOCK_CHILD_HOME"; + const CHILD_ATTEMPTING: &str = "ZED_PKG_TEST_STORE_LOCK_CHILD_ATTEMPTING"; + const CHILD_ACQUIRED: &str = "ZED_PKG_TEST_STORE_LOCK_CHILD_ACQUIRED"; + const TEST_NAME: &str = + "store::tests::install_process_lock_blocks_without_polling_until_owner_release"; + + if let Some(home) = std::env::var_os(CHILD_HOME) { + let attempting = PathBuf::from(std::env::var_os(CHILD_ATTEMPTING).unwrap()); + let acquired = PathBuf::from(std::env::var_os(CHILD_ACQUIRED).unwrap()); + fs::write(attempting, b"attempting").unwrap(); + let store = Store::new(Path::new(&home)); + let _waiter = store.install_lock().unwrap(); + fs::write(acquired, b"acquired").unwrap(); + return; + } + + let temp = tempfile::tempdir().unwrap(); + let store = Store::new(temp.path()); + let owner = store.install_lock().unwrap(); + let attempting = temp.path().join("store-lock-waiter-attempting"); + let acquired = temp.path().join("store-lock-waiter-acquired"); + let mut child = Command::new(std::env::current_exe().unwrap()) + .arg(TEST_NAME) + .arg("--exact") + .arg("--nocapture") + .env(CHILD_HOME, temp.path()) + .env(CHILD_ATTEMPTING, &attempting) + .env(CHILD_ACQUIRED, &acquired) + .stdin(Stdio::null()) + .stdout(Stdio::inherit()) + .stderr(Stdio::inherit()) + .spawn() + .unwrap(); + + let attempting_deadline = Instant::now() + Duration::from_secs(5); + while !attempting.is_file() { + if let Some(status) = child.try_wait().unwrap() { + panic!("store-lock waiter exited before attempting acquisition: {status}"); + } + if Instant::now() >= attempting_deadline { + let _ = child.kill(); + let _ = child.wait(); + panic!("store-lock waiter did not reach acquisition"); + } + std::thread::sleep(Duration::from_millis(20)); + } + + std::thread::sleep(Duration::from_millis(200)); + assert!( + !acquired.exists(), + "a separate process acquired install.lock while the owner held it" + ); + assert!( + child.try_wait().unwrap().is_none(), + "store-lock waiter exited before owner release" + ); + + drop(owner); + let release_deadline = Instant::now() + Duration::from_secs(5); + loop { + if let Some(status) = child.try_wait().unwrap() { + assert!(status.success(), "store-lock waiter failed: {status}"); + break; + } + if Instant::now() >= release_deadline { + let _ = child.kill(); + let _ = child.wait(); + panic!("store-lock waiter did not wake after owner release"); + } + std::thread::sleep(Duration::from_millis(20)); + } + assert!( + acquired.is_file(), + "waiter exited without acquiring install.lock" + ); + } + + #[test] + fn install_process_lock_is_released_when_owner_process_is_terminated() { + const CHILD_ROLE: &str = "ZED_PKG_TEST_STORE_LOCK_EXIT_ROLE"; + const CHILD_HOME: &str = "ZED_PKG_TEST_STORE_LOCK_EXIT_HOME"; + const OWNER_ACQUIRED: &str = "ZED_PKG_TEST_STORE_LOCK_OWNER_ACQUIRED"; + const WAITER_ATTEMPTING: &str = "ZED_PKG_TEST_STORE_LOCK_WAITER_ATTEMPTING"; + const WAITER_ACQUIRED: &str = "ZED_PKG_TEST_STORE_LOCK_WAITER_ACQUIRED"; + const TEST_NAME: &str = + "store::tests::install_process_lock_is_released_when_owner_process_is_terminated"; + + if let Some(role) = std::env::var_os(CHILD_ROLE) { + let home = PathBuf::from(std::env::var_os(CHILD_HOME).unwrap()); + let store = Store::new(&home); + match role.to_string_lossy().as_ref() { + "owner" => { + let _owner = store.install_lock().unwrap(); + fs::write(std::env::var_os(OWNER_ACQUIRED).unwrap(), b"acquired").unwrap(); + loop { + std::thread::sleep(Duration::from_secs(60)); + } + } + "waiter" => { + fs::write(std::env::var_os(WAITER_ATTEMPTING).unwrap(), b"attempting").unwrap(); + let _waiter = store.install_lock().unwrap(); + fs::write(std::env::var_os(WAITER_ACQUIRED).unwrap(), b"acquired").unwrap(); + return; + } + other => panic!("unexpected child role: {other}"), + } + } + + let temp = tempfile::tempdir().unwrap(); + let owner_acquired = temp.path().join("store-lock-owner-acquired"); + let waiter_attempting = temp.path().join("store-lock-waiter-attempting-after-exit"); + let waiter_acquired = temp.path().join("store-lock-waiter-acquired-after-exit"); + + let mut owner = Command::new(std::env::current_exe().unwrap()) + .arg(TEST_NAME) + .arg("--exact") + .arg("--nocapture") + .env(CHILD_ROLE, "owner") + .env(CHILD_HOME, temp.path()) + .env(OWNER_ACQUIRED, &owner_acquired) + .stdin(Stdio::null()) + .stdout(Stdio::inherit()) + .stderr(Stdio::inherit()) + .spawn() + .unwrap(); + + let owner_deadline = Instant::now() + Duration::from_secs(5); + while !owner_acquired.is_file() { + if let Some(status) = owner.try_wait().unwrap() { + panic!("store-lock owner exited before acquiring install.lock: {status}"); + } + if Instant::now() >= owner_deadline { + let _ = owner.kill(); + let _ = owner.wait(); + panic!("store-lock owner did not acquire install.lock"); + } + std::thread::sleep(Duration::from_millis(20)); + } + + let mut waiter = Command::new(std::env::current_exe().unwrap()) + .arg(TEST_NAME) + .arg("--exact") + .arg("--nocapture") + .env(CHILD_ROLE, "waiter") + .env(CHILD_HOME, temp.path()) + .env(WAITER_ATTEMPTING, &waiter_attempting) + .env(WAITER_ACQUIRED, &waiter_acquired) + .stdin(Stdio::null()) + .stdout(Stdio::inherit()) + .stderr(Stdio::inherit()) + .spawn() + .unwrap(); + + let attempting_deadline = Instant::now() + Duration::from_secs(5); + while !waiter_attempting.is_file() { + if let Some(status) = waiter.try_wait().unwrap() { + let _ = owner.kill(); + let _ = owner.wait(); + panic!("store-lock waiter exited before attempting acquisition: {status}"); + } + if Instant::now() >= attempting_deadline { + let _ = waiter.kill(); + let _ = waiter.wait(); + let _ = owner.kill(); + let _ = owner.wait(); + panic!("store-lock waiter did not reach acquisition"); + } + std::thread::sleep(Duration::from_millis(20)); + } + + std::thread::sleep(Duration::from_millis(200)); + assert!( + !waiter_acquired.exists(), + "waiter acquired install.lock while the owner process was alive" + ); + assert!( + waiter.try_wait().unwrap().is_none(), + "store-lock waiter exited before owner termination" + ); + + owner.kill().unwrap(); + let owner_status = owner.wait().unwrap(); + assert!( + !owner_status.success(), + "owner process unexpectedly exited successfully instead of being terminated" + ); + + let release_deadline = Instant::now() + Duration::from_secs(5); + loop { + if let Some(status) = waiter.try_wait().unwrap() { + assert!( + status.success(), + "store-lock waiter failed after owner exit: {status}" + ); + break; + } + if Instant::now() >= release_deadline { + let _ = waiter.kill(); + let _ = waiter.wait(); + panic!("store-lock waiter did not wake after owner process termination"); + } + std::thread::sleep(Duration::from_millis(20)); + } + assert!( + waiter_acquired.is_file(), + "waiter exited without acquiring install.lock after owner termination" + ); + } + #[test] fn gc_survives_hostile_max_age() { // A hostile age (`u64::MAX` seconds) would overflow naive epoch math diff --git a/tests/recursive_install_concurrency.rs b/tests/recursive_install_concurrency.rs new file mode 100644 index 00000000..a8b2042a --- /dev/null +++ b/tests/recursive_install_concurrency.rs @@ -0,0 +1,438 @@ +use std::collections::HashMap; +use std::fs; +use std::io::{BufRead, BufReader, Write}; +use std::net::{SocketAddr, TcpListener, TcpStream}; +use std::path::{Path, PathBuf}; +use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering}; +use std::sync::{Arc, Condvar, Mutex}; +use std::thread::{self, JoinHandle}; +use std::time::{Duration, Instant}; + +use anyhow::{Context, Result}; +use zed_cli::config::Config; +use zed_cli::install_graph::prefetch; +use zed_cli::pack::pack; +use zed_cli::store::Store; +use zed_interfaces::manifest::Manifest; +use zed_interfaces::registry::{self, PackageMetadata, VersionMetadata}; +use zed_interfaces::vcs::Vcs; + +const ORG: &str = "recursive-concurrency"; +const VERSION: &str = "1.0.0"; +const PACKAGE_COUNT: usize = 12; +const EXPECTED_CONCURRENCY: usize = 5; + +#[derive(Clone)] +struct Response { + content_type: &'static str, + body: Arc>, + artifact: bool, +} + +#[derive(Default)] +struct GateState { + arrived: usize, + open: bool, + timed_out: bool, +} + +struct ServerState { + responses: HashMap, + active_artifacts: AtomicUsize, + max_active_artifacts: AtomicUsize, + artifact_requests: AtomicUsize, + handler_failed: AtomicBool, + gate: Mutex, + gate_ready: Condvar, +} + +struct RegistryServer { + address: SocketAddr, + state: Arc, + shutdown: Arc, + accept_thread: Option>, +} + +impl RegistryServer { + fn start(listener: TcpListener, responses: HashMap) -> Result { + let address = listener.local_addr()?; + let state = Arc::new(ServerState { + responses, + active_artifacts: AtomicUsize::new(0), + max_active_artifacts: AtomicUsize::new(0), + artifact_requests: AtomicUsize::new(0), + handler_failed: AtomicBool::new(false), + gate: Mutex::new(GateState::default()), + gate_ready: Condvar::new(), + }); + let shutdown = Arc::new(AtomicBool::new(false)); + let accept_state = Arc::clone(&state); + let accept_shutdown = Arc::clone(&shutdown); + let accept_thread = thread::Builder::new() + .name("recursive-registry".to_string()) + .spawn(move || { + while let Ok((stream, _)) = listener.accept() { + if accept_shutdown.load(Ordering::SeqCst) { + break; + } + let handler_state = Arc::clone(&accept_state); + thread::spawn(move || { + if let Err(error) = handle_connection(stream, &handler_state) { + handler_state.handler_failed.store(true, Ordering::SeqCst); + eprintln!("test registry handler failed: {error:#}"); + } + }); + } + }) + .context("starting test registry accept loop")?; + Ok(Self { + address, + state, + shutdown, + accept_thread: Some(accept_thread), + }) + } + + fn base_url(&self) -> String { + format!("http://{}", self.address) + } + + fn max_active_artifacts(&self) -> usize { + self.state.max_active_artifacts.load(Ordering::SeqCst) + } + + fn artifact_requests(&self) -> usize { + self.state.artifact_requests.load(Ordering::SeqCst) + } + + fn gate_timed_out(&self) -> bool { + self.state + .gate + .lock() + .unwrap_or_else(|poisoned| poisoned.into_inner()) + .timed_out + } +} + +impl Drop for RegistryServer { + fn drop(&mut self) { + self.shutdown.store(true, Ordering::SeqCst); + let _ = TcpStream::connect(self.address); + if let Some(thread) = self.accept_thread.take() { + let _ = thread.join(); + } + } +} + +fn handle_connection(mut stream: TcpStream, state: &ServerState) -> Result<()> { + let mut reader = BufReader::new(stream.try_clone()?); + let mut request_line = String::new(); + if reader.read_line(&mut request_line)? == 0 { + return Ok(()); + } + let path = request_line + .split_whitespace() + .nth(1) + .context("HTTP request has a path")? + .split('?') + .next() + .unwrap_or_default() + .to_string(); + + loop { + let mut header = String::new(); + if reader.read_line(&mut header)? == 0 || header == "\r\n" || header == "\n" { + break; + } + } + + let Some(response) = state.responses.get(&path) else { + return write_response(&mut stream, "404 Not Found", "text/plain", b"not found\n"); + }; + + if !response.artifact { + return write_response( + &mut stream, + "200 OK", + response.content_type, + response.body.as_slice(), + ); + } + + state.artifact_requests.fetch_add(1, Ordering::SeqCst); + let active = state.active_artifacts.fetch_add(1, Ordering::SeqCst) + 1; + state + .max_active_artifacts + .fetch_max(active, Ordering::SeqCst); + + let mut gate = state + .gate + .lock() + .unwrap_or_else(|poisoned| poisoned.into_inner()); + gate.arrived += 1; + if gate.arrived >= EXPECTED_CONCURRENCY { + gate.open = true; + state.gate_ready.notify_all(); + } + let deadline = Instant::now() + Duration::from_secs(5); + while !gate.open { + let remaining = deadline.saturating_duration_since(Instant::now()); + if remaining.is_zero() { + gate.timed_out = true; + gate.open = true; + state.gate_ready.notify_all(); + break; + } + let (next, timeout) = state + .gate_ready + .wait_timeout(gate, remaining) + .unwrap_or_else(|poisoned| poisoned.into_inner()); + gate = next; + if timeout.timed_out() && !gate.open { + gate.timed_out = true; + gate.open = true; + state.gate_ready.notify_all(); + } + } + drop(gate); + + // Keep the first wave alive briefly so an implementation that exceeds the + // configured bound is visible in max_active_artifacts rather than hidden + // by very small fixture archives. + thread::sleep(Duration::from_millis(75)); + let result = write_response( + &mut stream, + "200 OK", + response.content_type, + response.body.as_slice(), + ); + state.active_artifacts.fetch_sub(1, Ordering::SeqCst); + result +} + +fn write_response( + stream: &mut TcpStream, + status: &str, + content_type: &str, + body: &[u8], +) -> Result<()> { + write!( + stream, + "HTTP/1.1 {status}\r\nContent-Type: {content_type}\r\nContent-Length: {}\r\nConnection: close\r\n\r\n", + body.len() + )?; + stream.write_all(body)?; + stream.flush()?; + Ok(()) +} + +struct PublishedPackage { + name: String, + package: PackageMetadata, + version: VersionMetadata, + artifact: Vec, +} + +fn package_manifest(name: &str) -> String { + format!( + r#"[package] +org = "{ORG}" +name = "{name}" +version = "{VERSION}" +description = "bounded recursive install fixture" + +[package.repository] +vcs = "git" +url = "https://example.invalid/{ORG}/{name}" +"#, + ) +} + +fn build_package(scratch: &Path, name: &str) -> Result { + let source = scratch.join(format!("source-{name}")); + fs::create_dir_all(&source)?; + let manifest_text = package_manifest(name); + fs::write(source.join(".zpkg.toml"), &manifest_text)?; + fs::write( + source.join("payload.txt"), + format!("{ORG}/{name}@{VERSION}\n"), + )?; + let manifest = Manifest::parse(&manifest_text)?; + let packed = pack( + &source, + &manifest, + Some(&scratch.join(format!("packed-{name}"))), + )?; + let artifact = fs::read(&packed.path)?; + Ok(PublishedPackage { + name: name.to_string(), + package: PackageMetadata { + org: ORG.to_string(), + name: name.to_string(), + description: Some(format!("fixture {name}")), + vcs: Vcs::Git, + repo_url: format!("https://example.invalid/{ORG}/{name}"), + version_scheme: manifest.package.version_scheme, + latest: Some(VERSION.to_string()), + tags: Vec::new(), + versions: vec![VERSION.to_string()], + }, + version: VersionMetadata { + org: ORG.to_string(), + name: name.to_string(), + version: VERSION.to_string(), + sha256: packed.sha256, + size: packed.size, + format: packed.format, + vcs_tag: format!("v{VERSION}"), + vcs_commit: Some("0123456789abcdef0123456789abcdef01234567".to_string()), + download_url: String::new(), + published_at: "1970-01-01T00:00:00Z".to_string(), + yanked: false, + }, + artifact, + }) +} + +fn registry_responses( + address: SocketAddr, + packages: &mut [PublishedPackage], +) -> Result> { + let mut responses = HashMap::new(); + for package in packages { + let artifact_path = registry::artifact_path(&package.version.sha256); + package.version.download_url = format!("http://{address}{artifact_path}"); + responses.insert( + registry::package_path(ORG, &package.name), + Response { + content_type: "application/json", + body: Arc::new(serde_json::to_vec(&package.package)?), + artifact: false, + }, + ); + responses.insert( + registry::version_path(ORG, &package.name, VERSION), + Response { + content_type: "application/json", + body: Arc::new(serde_json::to_vec(&package.version)?), + artifact: false, + }, + ); + responses.insert( + artifact_path, + Response { + content_type: "application/octet-stream", + body: Arc::new(package.artifact.clone()), + artifact: true, + }, + ); + } + Ok(responses) +} + +fn write_consumer(project: &Path, packages: &[PublishedPackage]) -> Result<()> { + fs::create_dir_all(project)?; + let mut manifest = String::from( + r#"[package] +org = "recursive-consumer" +name = "wide-frontier" +version = "0.1.0" + +[package.repository] +vcs = "git" +url = "https://example.invalid/recursive-consumer/wide-frontier" + +[dependencies] +"#, + ); + for package in packages { + manifest.push_str(&format!("\"{ORG}/{}\" = \"={VERSION}\"\n", package.name)); + } + fs::write(project.join(".zpkg.toml"), manifest)?; + Ok(()) +} + +fn test_config(registry: &str, home: PathBuf) -> Config { + Config { + registry: registry.to_string(), + home, + token: None, + auth_url: "http://127.0.0.1/unused".to_string(), + supabase_url: None, + supabase_key: None, + interactive: false, + } +} + +#[test] +fn recursive_http_prefetch_saturates_at_five_and_warm_runs_do_not_redownload() -> Result<()> { + if let Ok(raw) = std::env::var("ZED_PKG_INSTALL_CONCURRENCY") { + assert_eq!( + raw, "5", + "this contract test expects the default five-worker setting or an explicit value of 5" + ); + } + + let temp = tempfile::tempdir()?; + let scratch = temp.path().join("scratch"); + let project = temp.path().join("project"); + let home = temp.path().join("home"); + let listener = TcpListener::bind("127.0.0.1:0")?; + let address = listener.local_addr()?; + + let mut packages = (0..PACKAGE_COUNT) + .map(|index| build_package(&scratch, &format!("pkg-{index:02}"))) + .collect::>>()?; + write_consumer(&project, &packages)?; + let server = RegistryServer::start(listener, registry_responses(address, &mut packages)?)?; + let config = test_config(&server.base_url(), home.clone()); + + let cold = prefetch(&project, &config, false)?; + assert_eq!(cold.resolved, PACKAGE_COUNT); + assert_eq!(cold.downloaded, PACKAGE_COUNT); + assert_eq!(server.artifact_requests(), PACKAGE_COUNT); + assert_eq!(server.max_active_artifacts(), EXPECTED_CONCURRENCY); + assert!( + !server.gate_timed_out(), + "five artifact handlers never became active together" + ); + assert!( + !server.state.handler_failed.load(Ordering::SeqCst), + "the fixture HTTP registry reported a handler failure" + ); + + let warm = prefetch(&project, &config, false)?; + assert_eq!(warm.resolved, PACKAGE_COUNT); + assert_eq!(warm.downloaded, 0); + assert_eq!( + server.artifact_requests(), + PACKAGE_COUNT, + "a warm content-addressed store must not fetch artifact bytes again" + ); + + let store = Store::new(&home); + for package in &packages { + assert!( + store.has(&package.version.sha256), + "missing store entry for {}/{}", + package.version.org, + package.version.name + ); + } + + let artifact_locks = fs::read_dir(home.join("locks"))? + .filter_map(|entry| entry.ok()) + .filter(|entry| entry.file_name().to_string_lossy().starts_with("artifact-")) + .count(); + assert_eq!(artifact_locks, PACKAGE_COUNT); + + let cache_entries = fs::read_dir(home.join("cache"))?.collect::>>()?; + assert_eq!(cache_entries.len(), PACKAGE_COUNT); + assert!( + cache_entries + .iter() + .all(|entry| entry.file_type().is_ok_and(|kind| kind.is_file())), + "atomic download staging directories must be removed" + ); + Ok(()) +}