Lock download file to avoid races

See #TBD for details.

This is one of two possible implementations, the easier one is using a temporary directory or a unique filename to ensure processes don't collide. Using locks has less platform support, but OTOH means that we don't create stray temp files that don't get cleaned up (because we never know if one of those files is actually used by another process).

This does not fix #TBD completely. It only fixed the downloaded, parallel `rustup-init` calls still fail at the transaction stage.

This PR introduces a `LockedFile` primitive that's modeled after the uv type of the same name. It's currently very simple and locks forever. We can extend it with a timeout that makes rustup exit after a certain duration and print an error message, instead of an inscrutable waiting without any message.

Author: konstin

src/dist/download.rs

@@ -15,7 +15,7 @@ use url::Url;
 use crate::config::Cfg;
 use crate::dist::manifest::Manifest;
 use crate::dist::{Channel, DEFAULT_DIST_SERVER, ToolchainDesc, temp};
-use crate::download::{download_file, download_file_with_resume};
+use crate::download::{LockedFile, download_file, download_file_with_resume};
 use crate::errors::RustupError;
 use crate::process::Process;
 use crate::utils;
@@ -55,6 +55,18 @@ impl<'a> DownloadCfg<'a> {
         utils::ensure_dir_exists("Download Directory", self.download_dir)?;
         let target_file = self.download_dir.join(Path::new(hash));
 
+        let file_name = target_file
+            .file_name()
+            .map(|s| s.to_str().unwrap_or("_"))
+            .unwrap_or("_")
+            .to_owned();
+        let locked_file = target_file.with_file_name(file_name.clone() + ".lock");
+
+        let _locked_file = LockedFile::create(&locked_file)
+            .with_context(|| RustupError::CreateLockedFile(locked_file.clone()))?
+            .lock()
+            .with_context(|| RustupError::LockedFile(locked_file.clone()))?;
+
         if target_file.exists() {
             let cached_result = file_hash(&target_file)?;
             if hash == cached_result {
@@ -67,14 +79,7 @@ impl<'a> DownloadCfg<'a> {
             }
         }
 
-        let partial_file_path = target_file.with_file_name(
-            target_file
-                .file_name()
-                .map(|s| s.to_str().unwrap_or("_"))
-                .unwrap_or("_")
-                .to_owned()
-                + ".partial",
-        );
+        let partial_file_path = target_file.with_file_name(file_name + ".partial");
 
         let partial_file_existed = partial_file_path.exists();
 

src/download/mod.rs

@@ -1,8 +1,9 @@
 //! Easy file downloading
 
-use std::fs::remove_file;
+use std::fs::{File, OpenOptions, remove_file};
+use std::io;
 use std::num::NonZero;
-use std::path::Path;
+use std::path::{Path, PathBuf};
 use std::str::FromStr;
 use std::time::Duration;
 
@@ -26,6 +27,95 @@ use crate::{dist::download::DownloadStatus, errors::RustupError, process::Proces
 #[cfg(test)]
 mod tests;
 
+/// An OS lock on a file that unlocks when dropping the file.
+pub(crate) struct LockedFile {
+    path: PathBuf,
+    file: File,
+}
+
+impl LockedFile {
+    /// Creates the file if it does not exit, does not lock.
+    #[cfg(unix)]
+    pub(crate) fn create(path: impl Into<PathBuf>) -> Result<Self, io::Error> {
+        use std::os::unix::fs::PermissionsExt;
+        use tempfile::NamedTempFile;
+
+        let path = path.into();
+
+        // If path already exists, return it.
+        if let Ok(file) = OpenOptions::new().read(true).write(true).open(&path) {
+            return Ok(Self { path, file });
+        }
+
+        // Otherwise, create a temporary file with 777 permissions, to handle races between
+        // processes running under different UIDs (e.g., in Docker containers). We must set
+        // permissions _after_ creating the file, to override the `umask`.
+        let file = if let Some(parent) = path.parent() {
+            NamedTempFile::new_in(parent)?
+        } else {
+            NamedTempFile::new()?
+        };
+        if let Err(err) = file
+            .as_file()
+            .set_permissions(std::fs::Permissions::from_mode(0o777))
+        {
+            warn!("failed to set permissions on temporary file: {err}");
+        }
+
+        // Try to move the file to path, but if path exists now, just open path
+        match file.persist_noclobber(&path) {
+            Ok(file) => Ok(Self { path, file }),
+            Err(err) => {
+                if err.error.kind() == io::ErrorKind::AlreadyExists {
+                    let file = OpenOptions::new().read(true).write(true).open(&path)?;
+                    Ok(Self { path, file })
+                } else {
+                    Err(err.error)
+                }
+            }
+        }
+    }
+
+    /// Creates the file if it does not exit, does not lock.
+    #[cfg(not(unix))]
+    pub(crate) fn create(path: impl Into<Path>) -> Result<Self, io::Error> {
+        let path = path.into();
+        let file = OpenOptions::new()
+            .read(true)
+            .write(true)
+            .create(true)
+            .open(&path)?;
+        Ok(Self { path, file })
+    }
+
+    /// Acquire an exclusive lock on a file, blocking until the file is ready.
+    pub(crate) fn lock(self) -> Result<Self, io::Error> {
+        match self.file.lock() {
+            Err(err) if err.kind() == io::ErrorKind::Unsupported => {
+                warn!("locking files is not supported, running rustup in parallel may error");
+                Ok(self)
+            }
+            Err(err) => Err(err),
+            Ok(()) => Ok(self),
+        }
+    }
+}
+
+impl Drop for LockedFile {
+    /// Unlock the file.
+    fn drop(&mut self) {
+        if let Err(err) = self.file.unlock() {
+            warn!(
+                "failed to unlock {}; program may be stuck: {}",
+                self.path.display(),
+                err
+            );
+        } else {
+            debug!("released lock at `{}`", self.path.display());
+        }
+    }
+}
+
 pub(crate) async fn download_file(
     url: &Url,
     path: &Path,
@@ -48,7 +138,7 @@ pub(crate) async fn download_file_with_resume(
     match download_file_(url, path, hasher, resume_from_partial, status, process).await {
         Ok(_) => Ok(()),
         Err(e) => {
-            if e.downcast_ref::<std::io::Error>().is_some() {
+            if e.downcast_ref::<io::Error>().is_some() {
                 return Err(e);
             }
             let is_client_error = match e.downcast_ref::<DEK>() {
@@ -722,7 +812,7 @@ enum DownloadError {
     #[error("{0}")]
     Message(String),
     #[error(transparent)]
-    IoError(#[from] std::io::Error),
+    IoError(#[from] io::Error),
     #[cfg(any(feature = "reqwest-rustls-tls", feature = "reqwest-native-tls"))]
     #[error(transparent)]
     Reqwest(#[from] ::reqwest::Error),

src/errors.rs

@@ -27,6 +27,10 @@ pub struct OperationError(pub anyhow::Error);
 pub enum RustupError {
     #[error("partially downloaded file may have been damaged and was removed, please try again")]
     BrokenPartialFile,
+    #[error("failed to create '{0}' as lockfile")]
+    CreateLockedFile(PathBuf),
+    #[error("failed to lock '{0}'")]
+    LockedFile(PathBuf),
     #[error("component download failed for {0}")]
     ComponentDownloadFailed(String),
     #[error("failure removing component '{name}', directory does not exist: '{}'", .path.display())]