fix(tarball): use rwlock instead of dashmap to avoid potential deadlock

Cargo.toml

@@ -35,7 +35,6 @@ async-recursion    = { version = "1.0.5" }
 clap               = { version = "4", features = ["derive", "string"] }
 command-extra      = { version = "1.0.0" }
 base64             = { version = "0.21.5" }
-dashmap            = { version = "5.5.3" }
 derive_more        = { version = "1.0.0-beta.3", features = ["full"] }
 dunce              = { version = "1.0.4" }
 home               = { version = "0.5.5" }

crates/cli/src/state.rs

@@ -6,6 +6,7 @@ use pacquet_package_manifest::{PackageManifest, PackageManifestError};
 use pacquet_tarball::MemCache;
 use pipe_trait::Pipe;
 use reqwest::Client;
+use std::collections::HashMap;
 use std::path::PathBuf;
 
 /// Application state when running `pacquet run` or `pacquet install`.
@@ -44,7 +45,7 @@ impl State {
             lockfile: call_load_lockfile(config.lockfile, Lockfile::load_from_current_dir)
                 .map_err(InitStateError::LoadLockfile)?,
             http_client: Client::new(),
-            tarball_mem_cache: MemCache::new(),
+            tarball_mem_cache: MemCache::new(HashMap::new()),
         })
     }
 }

crates/cli/tests/install.rs

@@ -6,9 +6,10 @@ use command_extra::CommandExtra;
 use pacquet_testing_utils::{
     bin::{AddMockedRegistry, CommandTempCwd},
     fs::{get_all_files, get_all_folders, is_symlink_or_junction},
+    panic_after,
 };
 use pipe_trait::Pipe;
-use std::fs;
+use std::{fs, thread, time::Duration};
 
 #[test]
 fn should_install_dependencies() {
@@ -131,3 +132,32 @@ fn should_install_index_files() {
 
     drop((root, mock_instance)); // cleanup
 }
+
+#[test]
+fn should_install_duplicated_dependencies() {
+    let CommandTempCwd { pacquet, root, workspace, .. } =
+        CommandTempCwd::init().add_mocked_registry();
+
+    eprintln!("Creating package.json...");
+    let manifest_path = workspace.join("package.json");
+    let package_json_content = serde_json::json!({
+        "dependencies": {
+            "express": "4.18.2",
+        },
+        "devDependencies": {
+            "express": "4.18.2",
+        },
+    });
+    fs::write(manifest_path, package_json_content.to_string()).expect("write to package.json");
+
+    panic_after(Duration::from_secs(30), move || {
+        thread::sleep(Duration::from_millis(200));
+        eprintln!("Executing command...");
+        pacquet.with_arg("install").assert().success();
+        eprintln!("Make sure the package is installed");
+        assert!(is_symlink_or_junction(&workspace.join("node_modules/express")).unwrap());
+        assert!(workspace.join("node_modules/.pnpm/express@4.18.2").exists());
+    });
+
+    drop(root); // cleanup
+}

crates/tarball/Cargo.toml

@@ -16,7 +16,6 @@ pacquet-fs          = { workspace = true }
 pacquet-store-dir   = { workspace = true }
 
 base64       = { workspace = true }
-dashmap      = { workspace = true }
 derive_more  = { workspace = true }
 miette       = { workspace = true }
 pipe-trait   = { workspace = true }

crates/tarball/src/lib.rs

@@ -2,13 +2,13 @@ use std::{
     collections::HashMap,
     ffi::OsString,
     io::{Cursor, Read},
+    mem::drop,
     path::PathBuf,
     sync::Arc,
     time::UNIX_EPOCH,
 };
 
 use base64::{engine::general_purpose::STANDARD as BASE64_STD, Engine};
-use dashmap::DashMap;
 use derive_more::{Display, Error, From};
 use miette::Diagnostic;
 use pacquet_fs::file_mode;
@@ -88,7 +88,7 @@ pub enum CacheValue {
 /// Internal in-memory cache of tarballs.
 ///
 /// The key of this hashmap is the url of each tarball.
-pub type MemCache = DashMap<String, Arc<RwLock<CacheValue>>>;
+pub type MemCache = RwLock<HashMap<String, Arc<RwLock<CacheValue>>>>;
 
 #[instrument(skip(gz_data), fields(gz_data_len = gz_data.len()))]
 fn decompress_gzip(gz_data: &[u8], unpacked_size: Option<usize>) -> Result<Vec<u8>, TarballError> {
@@ -130,9 +130,9 @@ impl<'a> DownloadTarballToStore<'a> {
 
         // QUESTION: I see no copying from existing store_dir, is there such mechanism?
         // TODO: If it's not implemented yet, implement it
-
-        if let Some(cache_lock) = mem_cache.get(package_url) {
-            let notify = match &*cache_lock.write().await {
+        let mem_cache_reader = mem_cache.read().await;
+        if let Some(cache_lock) = mem_cache_reader.get(package_url) {
+            let notify = match &*cache_lock.read().await {
                 CacheValue::Available(cas_paths) => {
                     return Ok(Arc::clone(cas_paths));
                 }
@@ -146,13 +146,19 @@ impl<'a> DownloadTarballToStore<'a> {
             }
             unreachable!("Failed to get or compute tarball data for {package_url:?}");
         } else {
+            drop(mem_cache_reader);
             let notify = Arc::new(Notify::new());
             let cache_lock = notify
                 .pipe_ref(Arc::clone)
                 .pipe(CacheValue::InProgress)
                 .pipe(RwLock::new)
                 .pipe(Arc::new);
-            if mem_cache.insert(package_url.to_string(), Arc::clone(&cache_lock)).is_some() {
+            if mem_cache
+                .write()
+                .await
+                .insert(package_url.to_string(), Arc::clone(&cache_lock))
+                .is_some()
+            {
                 tracing::warn!(target: "pacquet::download", ?package_url, "Race condition detected when writing to cache");
             }
             let cas_paths = self.run_without_mem_cache().await?.pipe(Arc::new);

crates/testing-utils/src/lib.rs

@@ -1,2 +1,22 @@
+use std::{sync::mpsc, thread, time::Duration};
 pub mod bin;
 pub mod fs;
+
+pub fn panic_after<T, F>(d: Duration, f: F) -> T
+where
+    T: Send + 'static,
+    F: FnOnce() -> T,
+    F: Send + 'static,
+{
+    let (done_tx, done_rx) = mpsc::channel();
+    let handle = thread::spawn(move || {
+        let val = f();
+        done_tx.send(()).expect("Unable to send completion signal");
+        val
+    });
+
+    match done_rx.recv_timeout(d) {
+        Ok(_) => handle.join().expect("test panicked"),
+        Err(_) => panic!("test timeout"),
+    }
+}