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.lock

@@ -41,6 +41,7 @@ criterion = { version = "0.7", features = ["html_reports"] }
 crossterm = { version = "0.29.0", features = ["event-stream"] }
 directories = "6.0.0"
 flate2 = "1.0.35"
+fs4 = "0.13"
 futures-util = "0.3.31"
 hex = "0.4.3"
 httpmock = "0.7"

Cargo.toml

@@ -11,6 +11,7 @@ rust-version.workspace = true
 backon = { workspace = true }
 directories = { workspace = true }
 flate2 = { workspace = true }
+fs4 = { workspace = true }
 futures-util = { workspace = true }
 hex = { workspace = true }
 indoc = { workspace = true }

crates/vite_install/Cargo.toml

@@ -7,6 +7,7 @@ use std::{
     process::{ExitStatus, Stdio},
 };
 
+use fs4::fs_std::FileExt;
 use semver::{Version, VersionReq};
 use serde::{Deserialize, Serialize};
 use tokio::{fs::remove_dir_all, process::Command};
@@ -374,9 +375,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 exceptions 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_exclusive()?;
+    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,12 +408,13 @@ 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 {
+    match remove_dir_all(path.as_ref()).await {
         Ok(()) => Ok(()),
         Err(e) => {
             if e.kind() == std::io::ErrorKind::NotFound {
                 Ok(())
             } else {
+                tracing::error!("remove_dir_all_force path: {:?} error: {e:?}", path.as_ref());
                 Err(e)
             }
         }

crates/vite_install/src/package_manager.rs

@@ -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,35 @@ 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<T>(
+  tasks: (() => Promise<T>)[],
+  maxConcurrency = cpus().length,
+): Promise<T[]> {
+  const results: T[] = [];
+  const executing: Promise<void>[] = [];
+
+  for (const task of tasks) {
+    const promise = task().then((result) => {
+      results.push(result);
+      executing.splice(executing.indexOf(promise), 1);
+    });
+
+    executing.push(promise);
+
+    if (executing.length >= maxConcurrency) {
+      await Promise.race(executing);
+    }
+  }
+
+  await Promise.all(executing);
+  return results;
+}
+
 export async function snapTest() {
   const { positionals } = parseArgs({
     allowPositionals: true,
@@ -41,16 +70,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/snap-test.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) {