Add a 5 min default timeout for deadlocks (#16342)

When a process is running and another calls `uv cache clean` or `uv
cache prune` we currently deadlock - sometimes until the CI timeout
(https://github.com/astral-sh/setup-uv/issues/588). To avoid this, we
add a default 5 min timeout waiting for a lock. 5 min balances allowing
in-progress builds to finish, especially with larger native
dependencies, while also giving timely errors for deadlocks on (remote)
systems.

Commit 1 is a refactoring.

This branch also fixes a problem with the logging where acquired and
released resources currently mismatch:

```
DEBUG Acquired lock for `https://github.com/tqdm/tqdm`
DEBUG Using existing Git source `https://github.com/tqdm/tqdm`
DEBUG Released lock at `C:\Users\Konsti\AppData\Local\uv\cache\git-v0\locks\16bb813afef8edd2`
```

crates/uv-python/src/environment.rs

@@ -7,7 +7,7 @@ use owo_colors::OwoColorize;
 use tracing::debug;
 
 use uv_cache::Cache;
-use uv_fs::{LockedFile, Simplified};
+use uv_fs::{LockedFile, LockedFileError, Simplified};
 use uv_pep440::Version;
 use uv_preview::Preview;
 
@@ -312,7 +312,7 @@ impl PythonEnvironment {
     }
 
     /// Grab a file lock for the environment to prevent concurrent writes across processes.
-    pub async fn lock(&self) -> Result<LockedFile, std::io::Error> {
+    pub async fn lock(&self) -> Result<LockedFile, LockedFileError> {
         self.0.interpreter.lock().await
     }
 

crates/uv-python/src/interpreter.rs

@@ -18,7 +18,9 @@ use tracing::{debug, trace, warn};
 use uv_cache::{Cache, CacheBucket, CachedByTimestamp, Freshness};
 use uv_cache_info::Timestamp;
 use uv_cache_key::cache_digest;
-use uv_fs::{LockedFile, PythonExt, Simplified, write_atomic_sync};
+use uv_fs::{
+    LockedFile, LockedFileError, LockedFileMode, PythonExt, Simplified, write_atomic_sync,
+};
 use uv_install_wheel::Layout;
 use uv_pep440::Version;
 use uv_pep508::{MarkerEnvironment, StringVersion};
@@ -666,17 +668,28 @@ impl Interpreter {
     }
 
     /// Grab a file lock for the environment to prevent concurrent writes across processes.
-    pub async fn lock(&self) -> Result<LockedFile, io::Error> {
+    pub async fn lock(&self) -> Result<LockedFile, LockedFileError> {
         if let Some(target) = self.target() {
             // If we're installing into a `--target`, use a target-specific lockfile.
-            LockedFile::acquire(target.root().join(".lock"), target.root().user_display()).await
+            LockedFile::acquire(
+                target.root().join(".lock"),
+                LockedFileMode::Exclusive,
+                target.root().user_display(),
+            )
+            .await
         } else if let Some(prefix) = self.prefix() {
             // Likewise, if we're installing into a `--prefix`, use a prefix-specific lockfile.
-            LockedFile::acquire(prefix.root().join(".lock"), prefix.root().user_display()).await
+            LockedFile::acquire(
+                prefix.root().join(".lock"),
+                LockedFileMode::Exclusive,
+                prefix.root().user_display(),
+            )
+            .await
         } else if self.is_virtualenv() {
             // If the environment a virtualenv, use a virtualenv-specific lockfile.
             LockedFile::acquire(
                 self.sys_prefix.join(".lock"),
+                LockedFileMode::Exclusive,
                 self.sys_prefix.user_display(),
             )
             .await
@@ -684,6 +697,7 @@ impl Interpreter {
             // Otherwise, use a global lockfile.
             LockedFile::acquire(
                 env::temp_dir().join(format!("uv-{}.lock", cache_digest(&self.sys_executable))),
+                LockedFileMode::Exclusive,
                 self.sys_prefix.user_display(),
             )
             .await
@@ -1272,8 +1286,8 @@ mod tests {
 
     use crate::Interpreter;
 
-    #[test]
-    fn test_cache_invalidation() {
+    #[tokio::test]
+    async fn test_cache_invalidation() {
         let mock_dir = tempdir().unwrap();
         let mocked_interpreter = mock_dir.path().join("python");
         let json = indoc! {r##"
@@ -1334,7 +1348,7 @@ mod tests {
         }
     "##};
 
-        let cache = Cache::temp().unwrap().init().unwrap();
+        let cache = Cache::temp().unwrap().init().await.unwrap();
 
         fs::write(
             &mocked_interpreter,

crates/uv-python/src/managed.rs

@@ -16,7 +16,9 @@ use uv_preview::{Preview, PreviewFeatures};
 #[cfg(windows)]
 use windows::Win32::Storage::FileSystem::FILE_ATTRIBUTE_REPARSE_POINT;
 
-use uv_fs::{LockedFile, Simplified, replace_symlink, symlink_or_copy_file};
+use uv_fs::{
+    LockedFile, LockedFileError, LockedFileMode, Simplified, replace_symlink, symlink_or_copy_file,
+};
 use uv_platform::{Error as PlatformError, Os};
 use uv_platform::{LibcDetectionError, Platform};
 use uv_state::{StateBucket, StateStore};
@@ -38,6 +40,8 @@ pub enum Error {
     #[error(transparent)]
     Io(#[from] io::Error),
     #[error(transparent)]
+    LockedFile(#[from] LockedFileError),
+    #[error(transparent)]
     Download(#[from] DownloadError),
     #[error(transparent)]
     PlatformError(#[from] PlatformError),
@@ -123,7 +127,12 @@ impl ManagedPythonInstallations {
     /// Grab a file lock for the managed Python distribution directory to prevent concurrent access
     /// across processes.
     pub async fn lock(&self) -> Result<LockedFile, Error> {
-        Ok(LockedFile::acquire(self.root.join(".lock"), self.root.user_display()).await?)
+        Ok(LockedFile::acquire(
+            self.root.join(".lock"),
+            LockedFileMode::Exclusive,
+            self.root.user_display(),
+        )
+        .await?)
     }
 
     /// Prefer, in order: