test(pm): add downloader race repro workflow

Author: elrrrrrrr

.github/workflows/downloader-race-repro.yml

@@ -0,0 +1,64 @@
+name: downloader-race-repro
+
+on:
+  workflow_dispatch:
+  pull_request:
+    paths:
+      - ".github/workflows/downloader-race-repro.yml"
+      - "crates/pm/src/util/downloader.rs"
+
+permissions:
+  contents: read
+
+jobs:
+  linux-repro:
+    name: linux downloader race repro
+    runs-on: ubuntu-latest
+    timeout-minutes: 45
+    steps:
+      - uses: actions/checkout@v4
+
+      - name: Init git submodules
+        run: git submodule update --init --recursive --depth 1
+
+      - name: Setup Node
+        uses: actions/setup-node@v4
+        with:
+          node-version: 20
+          cache: npm
+
+      - name: Install Rust
+        uses: dtolnay/rust-toolchain@stable
+        with:
+          toolchain: nightly-2026-01-04
+
+      - name: Cache cargo
+        uses: Swatinem/rust-cache@v2
+        with:
+          shared-key: downloader-race-repro-linux
+
+      - name: Install dependencies
+        run: npm install
+
+      - name: Try to reproduce zero-length resolved cache
+        shell: bash
+        run: |
+          set -uo pipefail
+
+          test_name='downloader::tests::repro_download_can_leave_zero_file_if_racing_process_dies_after_resolved'
+          for attempt in $(seq 1 30); do
+            echo "::group::attempt ${attempt}"
+            cargo test -p utoo-pm "${test_name}" -- --ignored --nocapture
+            status=$?
+            echo "::endgroup::"
+
+            if [ "${status}" -eq 0 ]; then
+              echo "::notice::Reproduced zero-length package.json in resolved downloader cache on attempt ${attempt}."
+              exit 0
+            fi
+
+            echo "attempt ${attempt} did not hit the zero-length window"
+          done
+
+          echo "::error::Did not reproduce zero-length package.json after 30 attempts."
+          exit 1

crates/pm/src/util/downloader.rs

@@ -307,10 +307,17 @@ mod tests {
     use flate2::Compression;
     use flate2::write::GzEncoder;
     use mockito::Server;
+    use std::env;
     use std::io::Write;
+    use std::path::PathBuf;
+    use std::process::{Child, Command};
+    use std::sync::Arc;
+    use std::sync::atomic::{AtomicUsize, Ordering};
+    use std::time::{Duration, Instant};
     use tar::Builder;
     use tempfile::TempDir;
     use tokio::task;
+    use tokio::time::sleep;
 
     // Helper to create a simple tar.gz archive in memory
     fn create_tar_gz() -> Vec<u8> {
@@ -330,6 +337,80 @@ mod tests {
         encoder.finish().unwrap()
     }
 
+    fn create_package_tar_gz() -> (Vec<u8>, String) {
+        let package_json = r#"{"name":"race-pkg","version":"1.0.0"}"#.to_string();
+        let mut tar_data = Vec::new();
+        {
+            let mut tar = Builder::new(&mut tar_data);
+            append_tar_file(&mut tar, "package/package.json", package_json.as_bytes());
+
+            for index in 0..128 {
+                let content = format!(
+                    "file-{index}\n{}",
+                    "0123456789abcdefghijklmnopqrstuvwxyz".repeat(256)
+                );
+                append_tar_file(
+                    &mut tar,
+                    &format!("package/lib/file-{index}.txt"),
+                    content.as_bytes(),
+                );
+            }
+
+            tar.finish().unwrap();
+        }
+
+        let mut encoder = GzEncoder::new(Vec::new(), Compression::default());
+        encoder.write_all(&tar_data).unwrap();
+        (encoder.finish().unwrap(), package_json)
+    }
+
+    fn create_large_package_json_tar_gz() -> (Vec<u8>, usize) {
+        let package_json = format!(
+            r#"{{"name":"race-pkg","version":"1.0.0","padding":"{}"}}"#,
+            "x".repeat(64 * 1024 * 1024)
+        );
+        let package_json_len = package_json.len();
+        let mut tar_data = Vec::new();
+        {
+            let mut tar = Builder::new(&mut tar_data);
+            append_tar_file(&mut tar, "package/package.json", package_json.as_bytes());
+            tar.finish().unwrap();
+        }
+
+        let mut encoder = GzEncoder::new(Vec::new(), Compression::default());
+        encoder.write_all(&tar_data).unwrap();
+        (encoder.finish().unwrap(), package_json_len)
+    }
+
+    fn append_tar_file(tar: &mut Builder<&mut Vec<u8>>, path: &str, content: &[u8]) {
+        let mut header = tar::Header::new_gnu();
+        header.set_path(path).unwrap();
+        header.set_size(content.len() as u64);
+        header.set_mode(0o644);
+        header.set_cksum();
+        tar.append(&header, content).unwrap();
+    }
+
+    fn spawn_gated_download_helper(
+        test_binary: &Path,
+        url: &str,
+        dest: &Path,
+        ready_dir: &Path,
+        start_file: &Path,
+    ) -> Child {
+        Command::new(test_binary)
+            .arg("util::downloader::tests::download_process_helper")
+            .arg("--ignored")
+            .arg("--nocapture")
+            .env("UTOO_DOWNLOAD_HELPER", "1")
+            .env("UTOO_DOWNLOAD_HELPER_URL", url)
+            .env("UTOO_DOWNLOAD_HELPER_DEST", dest)
+            .env("UTOO_DOWNLOAD_HELPER_READY_DIR", ready_dir)
+            .env("UTOO_DOWNLOAD_HELPER_START_FILE", start_file)
+            .spawn()
+            .unwrap()
+    }
+
     #[tokio::test]
     async fn test_download_idempotent() {
         let tar_gz = create_tar_gz();
@@ -369,4 +450,229 @@ mod tests {
         assert_eq!(content, "hello world");
         _m.assert();
     }
+
+    #[tokio::test]
+    #[ignore]
+    async fn test_download_shared_dest_across_processes() {
+        let (tar_gz, package_json) = create_package_tar_gz();
+        let body = Arc::new(tar_gz);
+        let mut server = Server::new_async().await;
+        let child_count = 8;
+        let _m = server
+            .mock("GET", "/pkg.tgz")
+            .with_status(200)
+            .with_header("content-type", "application/gzip")
+            .with_chunked_body(move |writer| {
+                std::thread::sleep(Duration::from_millis(200));
+                for chunk in body.chunks(1024) {
+                    writer.write_all(chunk)?;
+                    std::thread::sleep(Duration::from_millis(2));
+                }
+                Ok(())
+            })
+            .expect_at_least(2)
+            .expect_at_most(child_count)
+            .create_async()
+            .await;
+
+        let url = format!("{}/pkg.tgz", server.url());
+        let temp_dir = TempDir::new().unwrap();
+        let dest = temp_dir.path().join("pkg");
+        let ready_dir = temp_dir.path().join("ready");
+        let start_file = temp_dir.path().join("start");
+        crate::fs::create_dir_all(&ready_dir).await.unwrap();
+        let test_binary = env::current_exe().unwrap();
+        let mut children = Vec::new();
+
+        for _ in 0..child_count {
+            children.push(spawn_gated_download_helper(
+                &test_binary,
+                &url,
+                &dest,
+                &ready_dir,
+                &start_file,
+            ));
+        }
+
+        wait_for_ready_files(&ready_dir, child_count, Duration::from_secs(5)).await;
+        crate::fs::write(&start_file, "").await.unwrap();
+
+        let deadline = Instant::now() + Duration::from_secs(20);
+        let mut failures = Vec::new();
+        for mut child in children {
+            loop {
+                match child.try_wait().unwrap() {
+                    Some(status) => {
+                        if !status.success() {
+                            failures.push(status.to_string());
+                        }
+                        break;
+                    }
+                    None if Instant::now() >= deadline => {
+                        let _ = child.kill();
+                        failures.push("timed out".to_string());
+                        break;
+                    }
+                    None => sleep(Duration::from_millis(25)).await,
+                }
+            }
+        }
+
+        assert!(failures.is_empty(), "download helpers failed: {failures:?}");
+        assert!(dest.join("_resolved").exists());
+
+        let content = crate::fs::read_to_string(dest.join("package/package.json"))
+            .await
+            .unwrap();
+        assert_eq!(content, package_json);
+
+        for index in [0, 31, 63, 127] {
+            let path = dest.join(format!("package/lib/file-{index}.txt"));
+            let metadata = crate::fs::metadata(&path).await.unwrap();
+            assert!(metadata.len() > 0, "{} was empty", path.display());
+        }
+
+        _m.assert();
+    }
+
+    #[tokio::test]
+    #[ignore]
+    async fn repro_download_can_leave_zero_file_if_racing_process_dies_after_resolved() {
+        let (tar_gz, expected_package_json_len) = create_large_package_json_tar_gz();
+        let body = Arc::new(tar_gz);
+        let request_index = Arc::new(AtomicUsize::new(0));
+        let mut server = Server::new_async().await;
+        let _m = server
+            .mock("GET", "/pkg.tgz")
+            .with_status(200)
+            .with_header("content-type", "application/gzip")
+            .with_chunked_body(move |writer| {
+                let index = request_index.fetch_add(1, Ordering::SeqCst);
+                if index == 1 {
+                    std::thread::sleep(Duration::from_millis(500));
+                }
+
+                for chunk in body.chunks(64 * 1024) {
+                    writer.write_all(chunk)?;
+                }
+                Ok(())
+            })
+            .expect(2)
+            .create_async()
+            .await;
+
+        let url = format!("{}/pkg.tgz", server.url());
+        let temp_dir = TempDir::new().unwrap();
+        let dest = temp_dir.path().join("pkg");
+        let package_json_path = dest.join("package/package.json");
+        let ready_dir = temp_dir.path().join("ready");
+        let start_file = temp_dir.path().join("start");
+        crate::fs::create_dir_all(&ready_dir).await.unwrap();
+        let test_binary = env::current_exe().unwrap();
+        let children = vec![
+            spawn_gated_download_helper(&test_binary, &url, &dest, &ready_dir, &start_file),
+            spawn_gated_download_helper(&test_binary, &url, &dest, &ready_dir, &start_file),
+        ];
+
+        wait_for_ready_files(&ready_dir, 2, Duration::from_secs(5)).await;
+        crate::fs::write(&start_file, "").await.unwrap();
+
+        wait_for_path(&dest.join("_resolved"), Duration::from_secs(5)).await;
+        wait_for_zero_len(&package_json_path, Duration::from_secs(15)).await;
+
+        for child in &children {
+            if child.id() != 0 {
+                let _ = Command::new("kill")
+                    .arg("-KILL")
+                    .arg(child.id().to_string())
+                    .status();
+            }
+        }
+
+        for mut child in children {
+            let _ = child.wait();
+        }
+
+        assert!(dest.join("_resolved").exists());
+        assert_eq!(
+            crate::fs::metadata(&package_json_path).await.unwrap().len(),
+            0,
+            "racing writer finished instead of being killed after truncating package.json; expected full size was {expected_package_json_len}"
+        );
+
+        _m.assert();
+    }
+
+    async fn wait_for_path(path: &Path, timeout: Duration) {
+        let deadline = Instant::now() + timeout;
+        while Instant::now() < deadline {
+            if path.exists() {
+                return;
+            }
+            sleep(Duration::from_millis(10)).await;
+        }
+        panic!("timed out waiting for {}", path.display());
+    }
+
+    async fn wait_for_ready_files(path: &Path, count: usize, timeout: Duration) {
+        let deadline = Instant::now() + timeout;
+        while Instant::now() < deadline {
+            let ready_count = std::fs::read_dir(path)
+                .map(|entries| entries.count())
+                .unwrap_or(0);
+            if ready_count >= count {
+                return;
+            }
+            sleep(Duration::from_millis(10)).await;
+        }
+        panic!(
+            "timed out waiting for {count} ready files in {}",
+            path.display()
+        );
+    }
+
+    async fn wait_for_zero_len(path: &Path, timeout: Duration) {
+        let deadline = Instant::now() + timeout;
+        while Instant::now() < deadline {
+            if let Ok(metadata) = crate::fs::metadata(path).await
+                && metadata.len() == 0
+            {
+                return;
+            }
+            sleep(Duration::from_millis(10)).await;
+        }
+        panic!("timed out waiting for zero-length {}", path.display());
+    }
+
+    #[test]
+    #[ignore]
+    fn download_process_helper() {
+        if env::var("UTOO_DOWNLOAD_HELPER").ok().as_deref() != Some("1") {
+            return;
+        }
+
+        let url = env::var("UTOO_DOWNLOAD_HELPER_URL").unwrap();
+        let dest = PathBuf::from(env::var("UTOO_DOWNLOAD_HELPER_DEST").unwrap());
+        if let Ok(ready_dir) = env::var("UTOO_DOWNLOAD_HELPER_READY_DIR") {
+            std::fs::create_dir_all(&ready_dir).unwrap();
+            std::fs::write(
+                PathBuf::from(ready_dir).join(std::process::id().to_string()),
+                b"ready",
+            )
+            .unwrap();
+
+            let start_file = PathBuf::from(env::var("UTOO_DOWNLOAD_HELPER_START_FILE").unwrap());
+            while !start_file.exists() {
+                std::thread::sleep(Duration::from_millis(10));
+            }
+        }
+        let runtime = tokio::runtime::Builder::new_multi_thread()
+            .enable_all()
+            .build()
+            .unwrap();
+
+        runtime.block_on(async {
+            download(&url, &dest).await.unwrap();
+        });
+    }
 }