chore(core): move file lock to rust

packages/nx/Cargo.toml

@@ -74,3 +74,5 @@ assert_fs = "1.0.10"
 # This is only used for unit tests
 swc_ecma_dep_graph = "0.109.1"
 tempfile = "3.13.0"
+# We only explicitly use tokio for async tests
+tokio = "1.38.0"

packages/nx/src/native/index.d.ts

@@ -13,6 +13,14 @@ export declare class ChildProcess {
   onOutput(callback: (message: string) => void): void
 }
 
+export declare class FileLock {
+  locked: boolean
+  constructor(lockFilePath: string)
+  lock(): void
+  unlock(): void
+  wait(): Promise<void>
+}
+
 export declare class HashPlanner {
   constructor(nxJson: NxJson, projectGraph: ExternalObject<ProjectGraph>)
   getPlans(taskIds: Array<string>, taskGraph: TaskGraph): Record<string, string[]>

packages/nx/src/native/native-bindings.js

@@ -362,6 +362,7 @@ if (!nativeBinding) {
 }
 
 module.exports.ChildProcess = nativeBinding.ChildProcess
+module.exports.FileLock = nativeBinding.FileLock
 module.exports.HashPlanner = nativeBinding.HashPlanner
 module.exports.ImportResult = nativeBinding.ImportResult
 module.exports.NxCache = nativeBinding.NxCache

packages/nx/src/native/utils/file_lock.rs

@@ -0,0 +1,128 @@
+use std::fs::{self, File};
+use std::io;
+use std::path::Path;
+use std::time::Duration;
+
+#[napi]
+#[derive(Clone)]
+pub struct FileLock {
+    #[napi]
+    pub locked: bool,
+
+    lock_file_path: String,
+}
+
+#[napi]
+impl FileLock {
+    #[napi(constructor)]
+    pub fn new(lock_file_path: String) -> Self {
+        let locked = Path::new(&lock_file_path).exists();
+        Self {
+            locked,
+            lock_file_path,
+        }
+    }
+
+    #[napi]
+    pub fn lock(&mut self) -> anyhow::Result<()> {
+        if self.locked {
+            anyhow::bail!("File {} is already locked", self.lock_file_path)
+        }
+
+        let _ = File::create(&self.lock_file_path)?;
+        self.locked = true;
+        Ok(())
+    }
+
+    #[napi]
+    pub fn unlock(&mut self) -> anyhow::Result<()> {
+        if !self.locked {
+            anyhow::bail!("File {} is not locked", self.lock_file_path)
+        }
+        fs::remove_file(&self.lock_file_path).or_else(|err| {
+            if err.kind() == io::ErrorKind::NotFound {
+                Ok(())
+            } else {
+                Err(err)
+            }
+        })?;
+        self.locked = false;
+        Ok(())
+    }
+
+    #[napi]
+    pub async fn wait(&self) -> Result<(), napi::Error> {
+        if !self.locked {
+            return Ok(());
+        }
+
+        loop {
+            if !self.locked || !Path::new(&self.lock_file_path).exists() {
+                break Ok(());
+            }
+            std::thread::sleep(Duration::from_millis(2));
+        }
+    }
+}
+
+// Ensure the lock file is removed when the FileLock is dropped
+impl Drop for FileLock {
+    fn drop(&mut self) {
+        if self.locked {
+            let _ = self.unlock();
+        }
+    }
+}
+
+#[cfg(test)]
+mod test {
+    use super::*;
+
+    use assert_fs::prelude::*;
+    use assert_fs::TempDir;
+
+    #[test]
+    fn test_new_lock() {
+        let tmp_dir = TempDir::new().unwrap();
+        let lock_file = tmp_dir.child("test_lock_file");
+        let lock_file_path = lock_file.path().to_path_buf();
+        let lock_file_path_str = lock_file_path.into_os_string().into_string().unwrap();
+        let mut file_lock = FileLock::new(lock_file_path_str);
+        assert_eq!(file_lock.locked, false);
+        let _ = file_lock.lock();
+        assert_eq!(file_lock.locked, true);
+        assert!(lock_file.exists());
+        let _ = file_lock.unlock();
+        assert_eq!(lock_file.exists(), false);
+    }
+
+    #[tokio::test]
+    async fn test_wait() {
+        let tmp_dir = TempDir::new().unwrap();
+        let lock_file = tmp_dir.child("test_lock_file");
+        let lock_file_path = lock_file.path().to_path_buf();
+        let lock_file_path_str = lock_file_path.into_os_string().into_string().unwrap();
+        let mut file_lock = FileLock::new(lock_file_path_str);
+        let _ = file_lock.lock();
+        let file_lock_clone = file_lock.clone();
+        let wait_fut = async move {
+            let _ = file_lock_clone.wait().await;
+        };
+        let _ = tokio::runtime::Runtime::new().unwrap().block_on(wait_fut);
+        assert_eq!(file_lock.locked, false);
+        assert_eq!(lock_file.exists(), false);
+    }
+
+    #[test]
+    fn test_drop() {
+        let tmp_dir = TempDir::new().unwrap();
+        let lock_file = tmp_dir.child("test_lock_file");
+        let lock_file_path = lock_file.path().to_path_buf();
+        let lock_file_path_str = lock_file_path.into_os_string().into_string().unwrap();
+        {
+            let mut file_lock = FileLock::new(lock_file_path_str.clone());
+            let _ = file_lock.lock();
+        }
+        assert_eq!(lock_file.exists(), false);
+    }
+}

packages/nx/src/native/utils/mod.rs

@@ -11,5 +11,6 @@ pub use normalize_trait::Normalize;
 #[cfg_attr(target_arch = "wasm32", path = "atomics/wasm.rs")]
 pub mod atomics;
 pub mod ci;
+pub mod file_lock;
 
 pub use atomics::*;

packages/nx/src/project-graph/project-graph.ts

@@ -35,9 +35,10 @@ import {
 } from './utils/retrieve-workspace-files';
 import { getPlugins } from './plugins/get-plugins';
 import { logger } from '../utils/logger';
-import { FileLock } from '../utils/file-lock';
+import { FileLock } from '../native';
 import { join } from 'path';
 import { workspaceDataDirectory } from '../utils/cache-directory';
+import { DelayedSpinner } from '../utils/delayed-spinner';
 
 /**
  * Synchronously reads the latest cached copy of the workspace's ProjectGraph.
@@ -275,12 +276,35 @@ export async function createProjectGraphAndSourceMapsAsync(
   performance.mark('create-project-graph-async:start');
 
   if (!daemonClient.enabled()) {
-    const lock = new FileLock(join(workspaceDataDirectory, 'project-graph'));
+    const lock = new FileLock(
+      join(workspaceDataDirectory, 'project-graph.lock')
+    );
+
+    function cleanupFileLock() {
+      try {
+        lock.unlock();
+      } catch {}
+    }
+
+    process.on('exit', cleanupFileLock);
+
     if (lock.locked) {
       logger.verbose(
         'Waiting for graph construction in another process to complete'
       );
+      const spinner = new DelayedSpinner(
+        'Waiting for graph construction in another process to complete'
+      );
       await lock.wait();
+      spinner.cleanup();
+
+      // Note: This will currently throw if any of the caches are missing...
+      // It would be nice if one of the processes that was waiting for the lock
+      // could pick up the slack and build the graph if it's missing, but
+      // we wouldn't want either of the below to happen:
+      // - All of the waiting processes to build the graph
+      // - Even one of the processes building the graph on a legitimate error
+
       const sourceMaps = readSourceMapsCache();
       if (!sourceMaps) {
         throw new Error(
@@ -316,10 +340,11 @@ export async function createProjectGraphAndSourceMapsAsync(
         'create-project-graph-async:start',
         'create-project-graph-async:end'
       );
-      lock.unlock();
       return res;
     } catch (e) {
       handleProjectGraphError(opts, e);
+    } finally {
+      lock.unlock();
     }
   } else {
     try {