feat(install): add file locking to prevent concurrent installation issues (#289)

Add file locking to prevent race conditions in package manager installation

Added file-based locking to ensure atomic operations during package manager downloads, preventing concurrent processes from conflicting when installing the same package manager version.

close #140

## Changes

- Added file-based locking using `fs4` crate to synchronize critical operations
- Added NFS filesystem detection to skip locking on NFS where it may be unreliable
- Improved error logging in `remove_dir_all_force` to provide better debugging information
- Added `RUST_BACKTRACE=1` to CLI E2E tests for better error diagnostics
- Added `RUST_*` to passthrough environment variables in tools utils

## Why

This change prevents race conditions when multiple processes attempt to install the same package manager version simultaneously. The file lock ensures that the remove and rename operations are performed atomically, avoiding conflicts between concurrent installation attempts.

Author: fengmk2

.github/workflows/ci.yml

@@ -172,7 +172,7 @@ jobs:
 
       - name: Run CLI E2E tests
         run: |
-          pnpm test
+          RUST_BACKTRACE=1 pnpm test
           git diff --exit-code
 
   install-e2e-test:

Cargo.toml

@@ -6,7 +6,7 @@ members = ["bench", "crates/*", "packages/cli/binding"]
 authors = ["Vite+ Authors"]
 edition = "2024"
 license = "BUSL-1.1"
-rust-version = "1.88.0"
+rust-version = "1.89.0"
 
 [workspace.lints.rust]
 absolute_paths_not_starting_with_crate = "warn"

crates/vite_install/src/package_manager.rs

@@ -374,9 +374,21 @@ async fn download_package_manager(
     tracing::debug!("Rename package dir to {}", bin_name);
     tokio::fs::rename(&target_dir_tmp.join("package"), &target_dir_tmp.join(&bin_name)).await?;
 
-    // check bin_file again, for the concurrent download cases
+    // Use a file-based lock to ensure atomicity of remove + rename operations
+    // This prevents DirectoryNotEmpty error when multiple processes/threads
+    // try to install the same package manager version concurrently.
+    // The lock is automatically skipped on NFS filesystems where locking is unreliable.
+    let lock_path = parent_dir.join(format!("{version}.lock"));
+    tracing::debug!("Acquire lock file: {:?}", lock_path);
+    let lock_file = File::create(lock_path.as_path())?;
+    // Acquire exclusive lock (blocks until available)
+    lock_file.lock()?;
+    tracing::debug!("Lock acquired: {:?}", lock_path);
+
+    // Check again after acquiring the lock, in case another thread completed
+    // the installation while we were downloading
     if is_exists_file(&bin_file)? {
-        tracing::debug!("bin_file already exists, skip rename");
+        tracing::debug!("bin_file already exists after lock acquisition, skip rename");
         return Ok(install_dir);
     }
 
@@ -395,16 +407,15 @@ async fn download_package_manager(
 /// Remove the directory and all its contents.
 /// Ignore the error if the directory is not found.
 async fn remove_dir_all_force(path: impl AsRef<Path>) -> Result<(), std::io::Error> {
-    match remove_dir_all(path).await {
-        Ok(()) => Ok(()),
-        Err(e) => {
-            if e.kind() == std::io::ErrorKind::NotFound {
-                Ok(())
-            } else {
-                Err(e)
-            }
+    let path = path.as_ref();
+    remove_dir_all(path).await.or_else(|e| {
+        if e.kind() == std::io::ErrorKind::NotFound {
+            Ok(())
+        } else {
+            tracing::error!("remove_dir_all_force path: {:?} error: {e:?}", path);
+            Err(e)
         }
-    }
+    })
 }
 
 /// Create shim files for the package manager.

packages/tools/src/__tests__/__snapshots__/utils.spec.ts.snap

@@ -26,6 +26,8 @@ exports[`replaceUnstableOutput() > replace ignore pnpm request warning log 1`] =
 Packages:"
 `;
 
+exports[`replaceUnstableOutput() > replace pnpm registry request error warning log 1`] = `"Progress: resolved"`;
+
 exports[`replaceUnstableOutput() > replace tsdown output 1`] = `
 "ℹ tsdown v<semver> powered by rolldown v<semver>
 ℹ entry: src/index.ts

packages/tools/src/__tests__/utils.spec.ts

@@ -144,6 +144,14 @@ https://registry.yarnpkg.com/testnpm2/-/testnpm2-1.0.0.tgz
     `;
     expect(replaceUnstableOutput(output.trim())).toMatchSnapshot();
   });
+
+  test('replace pnpm registry request error warning log', () => {
+    const output = `
+ WARN  GET https://registry.npmjs.org/test-vite-plus-install error (ECONNRESET). Will retry in 10 seconds. 2 retries left.
+Progress: resolved
+`;
+    expect(replaceUnstableOutput(output.trim())).toMatchSnapshot();
+  });
 });
 
 describe('isPassThroughEnv()', () => {

packages/tools/src/snap-test.ts

@@ -2,7 +2,7 @@ import cp from 'node:child_process';
 import { randomUUID } from 'node:crypto';
 import fs from 'node:fs';
 import fsPromises from 'node:fs/promises';
-import { tmpdir } from 'node:os';
+import { cpus, tmpdir } from 'node:os';
 import path from 'node:path';
 import { debuglog, parseArgs, promisify } from 'node:util';
 
@@ -16,6 +16,44 @@ const exec = async (command: string, options: cp.ExecOptionsWithStringEncoding)
     process.platform === 'win32' ? { ...options, shell: 'pwsh.exe' } : options,
   );
 
+/**
+ * Run tasks with limited concurrency based on CPU count.
+ * @param tasks Array of task functions to execute
+ * @param maxConcurrency Maximum number of concurrent tasks (defaults to CPU count)
+ */
+async function runWithConcurrencyLimit(
+  tasks: (() => Promise<void>)[],
+  maxConcurrency = cpus().length,
+): Promise<void> {
+  const executing: Promise<void>[] = [];
+  const errors: Error[] = [];
+
+  for (const task of tasks) {
+    const promise = task()
+      .catch((error) => {
+        errors.push(error);
+        console.error('Task failed:', error);
+      })
+      .finally(() => {
+        executing.splice(executing.indexOf(promise), 1);
+      });
+
+    executing.push(promise);
+
+    if (executing.length >= maxConcurrency) {
+      await Promise.race(executing);
+    }
+  }
+
+  await Promise.all(executing);
+
+  if (errors.length > 0) {
+    throw new Error(
+      `${errors.length} test case(s) failed. First error: ${errors[0].message}`,
+    );
+  }
+}
+
 export async function snapTest() {
   const { positionals } = parseArgs({
     allowPositionals: true,
@@ -41,16 +79,22 @@ export async function snapTest() {
 
   const casesDir = path.resolve('snap-tests');
 
-  const tasks: Promise<void>[] = [];
+  const taskFunctions: (() => Promise<void>)[] = [];
   for (const caseName of fs.readdirSync(casesDir)) {
     if (caseName.startsWith('.')) continue; // Skip hidden files like .DS_Store
     if (caseName.includes(filter)) {
-      tasks.push(runTestCase(caseName, tempTmpDir, casesDir));
+      taskFunctions.push(() => runTestCase(caseName, tempTmpDir, casesDir));
     }
   }
 
-  if (tasks.length > 0) {
-    await Promise.all(tasks);
+  if (taskFunctions.length > 0) {
+    const cpuCount = cpus().length;
+    console.log(
+      'Running %d test cases with concurrency limit of %d (CPU count)',
+      taskFunctions.length,
+      cpuCount,
+    );
+    await runWithConcurrencyLimit(taskFunctions, cpuCount);
   }
 }
 

packages/tools/src/utils.ts

@@ -35,6 +35,8 @@ export function replaceUnstableOutput(output: string, cwd?: string) {
     .replaceAll(/ ?WARN\s+Request\s+took .+?\n/g, '')
     .replaceAll(/Scope: all \d+ workspace projects/g, 'Scope: all <variable> workspace projects')
     .replaceAll(/\++\n/g, '+<repeat>\n')
+    // ignore pnpm registry request error warning log
+    .replaceAll(/ ?WARN\s+GET\s+https:\/\/registry\..+?\n/g, '')
     // ignore yarn YN0013, because it's unstable output, only exists on CI environment
     // ➤ YN0013: │ A package was added to the project (+ 0.7 KiB).
     .replaceAll(/➤ YN0013:[^\n]+\n/g, '')
@@ -126,6 +128,8 @@ const DEFAULT_PASSTHROUGH_ENVS = [
   '*_TOKEN',
   // oxc specific
   'OXLINT_*',
+  // Rust specific
+  'RUST_*',
 ].map(env => new Minimatch(env));
 
 export function isPassThroughEnv(env: string) {