fix: lock to ensure atomicity of remove + rename operations

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/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

@@ -126,6 +126,8 @@ const DEFAULT_PASSTHROUGH_ENVS = [
   '*_TOKEN',
   // oxc specific
   'OXLINT_*',
+  // Rust specific
+  'RUST_*',
 ].map(env => new Minimatch(env));
 
 export function isPassThroughEnv(env: string) {