Bound concurrent Git diffs for untracked files (#54032)

## Why Collecting a diff previously launched a Git process for every untracked file at once, allowing process counts to grow with the number of files. ## What changed Limit untracked-file diffs to eight concurrent operations per collection while preserving patch order and skipping failed file diffs. Concurrent collections each retain their own limit. ## Testing Add tests for independent concurrency limits, failed-file handling, and patch order and application with tracked changes, binary files, and filenames containing spaces or a leading dash. GitOrigin-RevId: 896e896eae62ba9b7acf3e8946c235fe725ced40
2026-10-19 16:17:49 +00:00 · 2026-10-14 16:48:00 +00:00
parent 81aee8c86b
commit 8cbce63387
2 changed files with 274 additions and 7 deletions
@@ -0,0 +1,259 @@
+//! Verifies patch compatibility and bounded, independent untracked-diff collections.
+
+use super::diff_against_sha;
+use crate::GitSha;
+use codex_utils_process::background_command;
+use pretty_assertions::assert_eq;
+use std::path::Path;
+
+fn git(cwd: &Path, args: &[&str]) {
+    let output = background_command("git")
+        .args(args)
+        .current_dir(cwd)
+        .output()
+        .expect("run Git");
+    assert!(output.status.success(), "Git {args:?} failed: {output:?}");
+}
+
+#[tokio::test]
+async fn tracked_and_untracked_patches_preserve_order_and_apply() {
+    let repo = tempfile::tempdir().expect("create repository");
+    git(repo.path(), &["init", "-q"]);
+    git(repo.path(), &["config", "core.autocrlf", "false"]);
+    git(repo.path(), &["config", "core.fsmonitor", "false"]);
+    std::fs::write(repo.path().join("tracked.txt"), "before\n").expect("write tracked file");
+    git(repo.path(), &["add", "tracked.txt"]);
+    git(
+        repo.path(),
+        &[
+            "-c",
+            "user.name=Codex Tests",
+            "-c",
+            "user.email=codex-tests@example.com",
+            "-c",
+            "commit.gpgsign=false",
+            "-c",
+            if cfg!(windows) {
+                "core.hooksPath=NUL"
+            } else {
+                "core.hooksPath=/dev/null"
+            },
+            "commit",
+            "-qm",
+            "initial",
+        ],
+    );
+    let files: [(&str, &[u8]); 4] = [
+        ("tracked.txt", b"after\n"),
+        ("-new.txt", b"leading dash\n"),
+        ("a space.txt", b"space in filename\n"),
+        ("z.bin", b"\x00\x01\xffbinary\x00\xfe"),
+    ];
+    for (name, contents) in files {
+        std::fs::write(repo.path().join(name), contents).expect("write changed file");
+    }
+    let cwd = repo.path().to_path_buf();
+    // The production caller spawns collection; this also checks its Send bound.
+    let patch = tokio::spawn(async move {
+        diff_against_sha(Path::new("git"), &cwd, &GitSha::new("HEAD")).await
+    })
+    .await
+    .expect("join collection")
+    .expect("collect patch");
+    assert_eq!(
+        patch
+            .lines()
+            .filter(|line| line.starts_with("diff --git "))
+            .collect::<Vec<_>>(),
+        vec![
+            "diff --git a/tracked.txt b/tracked.txt",
+            "diff --git a/-new.txt b/-new.txt",
+            "diff --git a/a space.txt b/a space.txt",
+            "diff --git a/z.bin b/z.bin",
+        ]
+    );
+    assert!(patch.contains("GIT binary patch"));
+    git(repo.path(), &["reset", "--hard", "-q", "HEAD"]);
+    for (name, _) in &files[1..] {
+        std::fs::remove_file(repo.path().join(name)).expect("remove untracked file");
+    }
+    let patch_file = tempfile::NamedTempFile::new().expect("create patch file");
+    std::fs::write(patch_file.path(), patch).expect("save patch");
+    git(
+        repo.path(),
+        &[
+            "apply",
+            "--binary",
+            patch_file.path().to_str().expect("patch path"),
+        ],
+    );
+    assert_eq!(
+        files.map(|(name, _)| std::fs::read(repo.path().join(name)).expect("read applied file")),
+        files.map(|(_, contents)| contents.to_vec())
+    );
+}
+
+#[cfg(target_os = "linux")]
+struct GatedGit {
+    directory: tempfile::TempDir,
+    gates: Vec<std::fs::File>,
+    ready: std::fs::File,
+    events: futures::channel::mpsc::UnboundedReceiver<String>,
+    reader: Option<std::thread::JoinHandle<()>>,
+}
+
+#[cfg(target_os = "linux")]
+impl GatedGit {
+    fn new(file_counts: &[usize]) -> Self {
+        use std::io::BufRead;
+
+        let directory = tempfile::tempdir().expect("create fixture directory");
+        let ready_path = directory.path().join("ready");
+        let mut paths = vec![ready_path];
+        for (batch, count) in file_counts.iter().enumerate() {
+            let cwd = directory.path().join(format!("batch-{batch}"));
+            std::fs::create_dir(&cwd).expect("create batch directory");
+            paths.extend((0..*count).map(|index| cwd.join(format!("file-{index:02}"))));
+        }
+        let output = background_command("mkfifo")
+            .args(&paths)
+            .output()
+            .expect("create FIFO gates");
+        assert!(output.status.success(), "mkfifo failed: {output:?}");
+        let mut gates: Vec<_> = paths
+            .iter()
+            .map(|path| {
+                // Linux permits O_RDWR on a FIFO without another opener, so cleanup
+                // can always release gates even when their child has not started.
+                std::fs::OpenOptions::new()
+                    .read(true)
+                    .write(true)
+                    .open(path)
+                    .expect("open FIFO")
+            })
+            .collect();
+        let ready = gates.remove(0);
+        let input = ready.try_clone().expect("clone readiness FIFO");
+        let (sender, events) = futures::channel::mpsc::unbounded();
+        codex_utils_cargo_bin::write_executable(
+            &directory.path().join("git"),
+            r#"#!/bin/sh
+while [ "$1" = '-c' ]; do shift 2; done
+case "$1" in
+config) exit 1 ;;
+ls-files) for file in file-*; do printf '%s\n' "$file"; done ;;
+diff)
+  case " $* " in
+  *' --no-index '*)
+    for file do :; done
+    printf '%s/%s\n' "$PWD" "$file" > ../ready
+    read -r release < "$file"
+    if [ "$file" = 'file-04' ]; then printf 'failed-output\n'; exit 2; fi
+    printf '%s\n' "$file"
+    exit 1 ;;
+  *) printf 'tracked\n' ;;
+  esac ;;
+esac
+"#,
+        )
+        .expect("write fake Git");
+        let reader = std::thread::spawn(move || {
+            for line in std::io::BufReader::new(input).lines() {
+                let Ok(line) = line else { break };
+                if line == "stop" || sender.unbounded_send(line).is_err() {
+                    break;
+                }
+            }
+        });
+        Self {
+            directory,
+            gates,
+            ready,
+            events,
+            reader: Some(reader),
+        }
+    }
+
+    async fn ready(&mut self, batches: usize) {
+        use futures::StreamExt;
+
+        for _ in 0..8 * batches {
+            tokio::time::timeout(std::time::Duration::from_secs(3), self.events.next())
+                .await
+                .expect("each collection must fill its own eight slots")
+                .expect("read readiness event");
+        }
+    }
+
+    fn children(&self) -> usize {
+        use std::os::unix::ffi::OsStrExt;
+
+        let wrapper = self.directory.path().join("git");
+        let mut children = std::collections::BTreeSet::new();
+        for task in std::fs::read_dir("/proc/self/task").expect("list test threads") {
+            let task = task.expect("read thread entry");
+            if let Ok(pids) = std::fs::read_to_string(task.path().join("children")) {
+                children.extend(pids.split_whitespace().map(str::to_string));
+            }
+        }
+        children
+            .into_iter()
+            .filter(|pid| {
+                std::fs::read(format!("/proc/{pid}/cmdline")).is_ok_and(|args| {
+                    args.split(|byte| *byte == 0)
+                        .any(|arg| arg == wrapper.as_os_str().as_bytes())
+                })
+            })
+            .count()
+    }
+
+    fn release(&mut self) {
+        use std::io::Write;
+        for gate in &mut self.gates {
+            let _ = gate.write_all(b"go\n");
+        }
+    }
+}
+
+#[cfg(target_os = "linux")]
+impl Drop for GatedGit {
+    fn drop(&mut self) {
+        use std::io::Write;
+        self.release();
+        let _ = self.ready.write_all(b"stop\n");
+        if let Some(reader) = self.reader.take() {
+            reader.join().expect("join readiness reader");
+        }
+    }
+}
+
+#[cfg(target_os = "linux")]
+#[tokio::test]
+async fn untracked_collections_bound_children_independently_and_skip_failed_files() {
+    // At most 16 Git children even if the bound regresses. Keeping both phases
+    // in one test also prevents the fixtures from overlapping under nextest.
+    for counts in [&[12][..], &[8, 8][..]] {
+        let mut fixture = GatedGit::new(counts);
+        let collections: Vec<_> = (0..counts.len())
+            .map(|batch| {
+                let git = fixture.directory.path().join("git");
+                let cwd = fixture.directory.path().join(format!("batch-{batch}"));
+                tokio::spawn(
+                    async move { diff_against_sha(&git, &cwd, &GitSha::new("HEAD")).await },
+                )
+            })
+            .collect();
+        fixture.ready(counts.len()).await;
+        let children = fixture.children();
+        fixture.release();
+        for (collection, count) in collections.into_iter().zip(counts) {
+            let expected = "tracked\n".to_string()
+                + &(0..*count)
+                    .filter(|index| *index != 4)
+                    .map(|index| format!("file-{index:02}\n"))
+                    .collect::<String>();
+            assert_eq!(collection.await.expect("join collection"), Some(expected));
+        }
+        assert_eq!(children, 8 * counts.len());
+    }
+}
@@ -3,7 +3,8 @@ use std::collections::HashSet;
 use std::path::Path;
 use std::path::PathBuf;

-use futures::future::join_all;
+use futures::StreamExt;
+use futures::stream;
 use schemars::JsonSchema;
 use serde::Deserialize;
 use serde::Serialize;
@@ -39,6 +40,7 @@ pub fn get_git_repo_root(base_dir: &Path) -> Option<PathBuf> {
 /// Timeout for git commands to prevent freezing on large repositories
 const GIT_COMMAND_TIMEOUT: TokioDuration = TokioDuration::from_secs(5);
 const DISABLED_HOOKS_PATH: &str = if cfg!(windows) { "NUL" } else { "/dev/null" };
+const MAX_CONCURRENT_UNTRACKED_DIFFS: usize = 8;

 #[derive(Serialize, Deserialize, Clone, Debug, JsonSchema, TS)]
 pub struct GitInfo {
@@ -359,7 +361,7 @@ pub async fn git_diff_to_remote(cwd: &Path) -> Option<GitDiffToRemote> {
    let remotes = get_git_remotes(cwd).await?;
    let branches = branch_ancestry(cwd).await?;
    let base_sha = find_closest_sha(cwd, &branches, &remotes).await?;
-    let diff = diff_against_sha(cwd, &base_sha).await?;
+    let diff = diff_against_sha(Path::new("git"), cwd, &base_sha).await?;

    Some(GitDiffToRemote {
        sha: base_sha,
@@ -704,8 +706,7 @@ async fn find_closest_sha(cwd: &Path, branches: &[String], remotes: &[String]) -
    closest_sha.map(|(sha, _)| sha)
 }

-async fn diff_against_sha(cwd: &Path, sha: &GitSha) -> Option<String> {
-    let git = Path::new("git");
+async fn diff_against_sha(git: &Path, cwd: &Path, sha: &GitSha) -> Option<String> {
    let fsmonitor = detect_local_fsmonitor_override(git, cwd).await;
    let output = run_git_command_with_timeout_from(
        git,
@@ -756,9 +757,12 @@ async fn diff_against_sha(cwd: &Path, sha: &GitSha) -> Option<String> {
                ];
                run_git_command_with_timeout_from(git, &args_vec, cwd, fsmonitor).await
            });
-            let results = join_all(futures_iter).await;
-            for extra in results.into_iter().flatten() {
-                if extra.status.code().is_some_and(|c| c == 0 || c == 1)
+            // Keep earlier children reading and finishing while more paths remain.
+            let results = stream::iter(futures_iter).buffered(MAX_CONCURRENT_UNTRACKED_DIFFS);
+            futures::pin_mut!(results);
+            while let Some(extra) = results.next().await {
+                if let Some(extra) = extra
+                    && extra.status.code().is_some_and(|c| c == 0 || c == 1)
                    && let Ok(s) = String::from_utf8(extra.stdout)
                {
                    diff.push_str(&s);
@@ -770,6 +774,10 @@ async fn diff_against_sha(cwd: &Path, sha: &GitSha) -> Option<String> {
    Some(diff)
 }

+#[cfg(test)]
+#[path = "diff_tests.rs"]
+mod diff_tests;
+
 fn find_ancestor_git_entry(base_dir: &Path) -> Option<(PathBuf, PathBuf)> {
    let mut dir = base_dir.to_path_buf();