Split run concurrency knobs

crates/prek-consts/Cargo.toml

@@ -10,5 +10,8 @@ license = { workspace = true }
 [dependencies]
 tracing = { workspace = true }
 
+[features]
+test-utils = []
+
 [lints]
 workspace = true

crates/prek-consts/src/env_vars.rs

@@ -4,6 +4,54 @@ use tracing::info;
 
 pub struct EnvVars;
 
+pub trait EnvVarsRead {
+    fn var_os(&self, name: &str) -> Option<OsString>;
+
+    fn is_set(&self, name: &str) -> bool {
+        self.var_os(name).is_some()
+    }
+
+    fn var(&self, name: &str) -> Result<String, std::env::VarError> {
+        match self.var_os(name) {
+            Some(s) => s.into_string().map_err(std::env::VarError::NotUnicode),
+            None => Err(std::env::VarError::NotPresent),
+        }
+    }
+}
+
+impl EnvVarsRead for EnvVars {
+    fn var_os(&self, name: &str) -> Option<OsString> {
+        Self::var_os(name)
+    }
+}
+
+#[cfg(any(test, feature = "test-utils"))]
+#[derive(Debug, Clone, Copy)]
+pub struct EnvVarsMap<'a> {
+    values: &'a [(&'a str, &'a str)],
+}
+
+#[cfg(any(test, feature = "test-utils"))]
+impl EnvVarsMap<'_> {
+    fn direct_var_os(&self, name: &str) -> Option<OsString> {
+        self.values
+            .iter()
+            .find_map(|(key, value)| (*key == name).then(|| OsString::from(value)))
+    }
+}
+
+#[cfg(any(test, feature = "test-utils"))]
+impl EnvVarsRead for EnvVarsMap<'_> {
+    fn var_os(&self, name: &str) -> Option<OsString> {
+        self.direct_var_os(name).or_else(|| {
+            let name = EnvVars::pre_commit_name(name)?;
+            let val = self.direct_var_os(name)?;
+            info!("Falling back to pre-commit environment variable for {name}");
+            Some(val)
+        })
+    }
+}
+
 impl EnvVars {
     pub const PATH: &'static str = "PATH";
     pub const HOME: &'static str = "HOME";
@@ -23,6 +71,8 @@ impl EnvVars {
     pub const PREK_SKIP: &'static str = "PREK_SKIP";
     pub const PREK_ALLOW_NO_CONFIG: &'static str = "PREK_ALLOW_NO_CONFIG";
     pub const PREK_NO_CONCURRENCY: &'static str = "PREK_NO_CONCURRENCY";
+    pub const PREK_CONCURRENT_HOOKS: &'static str = "PREK_CONCURRENT_HOOKS";
+    pub const PREK_CONCURRENT_BATCHES: &'static str = "PREK_CONCURRENT_BATCHES";
     pub const PREK_MAX_CONCURRENCY: &'static str = "PREK_MAX_CONCURRENCY";
     pub const PREK_NO_FAST_PATH: &'static str = "PREK_NO_FAST_PATH";
     pub const PREK_UV_SOURCE: &'static str = "PREK_UV_SOURCE";
@@ -124,6 +174,13 @@ impl EnvVars {
     pub const DOTNET_ROOT: &'static str = "DOTNET_ROOT";
 }
 
+#[cfg(any(test, feature = "test-utils"))]
+impl EnvVars {
+    pub const fn from_map<'a>(values: &'a [(&'a str, &'a str)]) -> EnvVarsMap<'a> {
+        EnvVarsMap { values }
+    }
+}
+
 impl EnvVars {
     // Pre-commit environment variables that we support for compatibility
     pub const PRE_COMMIT_HOME: &'static str = "PRE_COMMIT_HOME";
@@ -205,7 +262,7 @@ impl EnvVars {
 
 #[cfg(test)]
 mod tests {
-    use super::EnvVars;
+    use super::{EnvVars, EnvVarsRead};
 
     #[test]
     fn test_parse_boolish() {
@@ -223,4 +280,28 @@ mod tests {
         assert_eq!(EnvVars::parse_boolish(""), None);
         assert_eq!(EnvVars::parse_boolish("123"), None);
     }
+
+    #[test]
+    fn test_env_vars_map_reads_values() {
+        let env_vars = EnvVars::from_map(&[(EnvVars::PREK_COLOR, "never")]);
+
+        assert_eq!(env_vars.var(EnvVars::PREK_COLOR).unwrap(), "never");
+    }
+
+    #[test]
+    fn test_env_vars_map_falls_back_to_pre_commit_name() {
+        let env_vars = EnvVars::from_map(&[(EnvVars::PRE_COMMIT_NO_CONCURRENCY, "1")]);
+
+        assert_eq!(env_vars.var(EnvVars::PREK_NO_CONCURRENCY).unwrap(), "1");
+    }
+
+    #[test]
+    fn test_env_vars_map_prefers_prek_name_over_pre_commit_name() {
+        let env_vars = EnvVars::from_map(&[
+            (EnvVars::PREK_NO_CONCURRENCY, "prek"),
+            (EnvVars::PRE_COMMIT_NO_CONCURRENCY, "pre-commit"),
+        ]);
+
+        assert_eq!(env_vars.var(EnvVars::PREK_NO_CONCURRENCY).unwrap(), "prek");
+    }
 }

crates/prek/Cargo.toml

@@ -108,6 +108,7 @@ etcetera = { workspace = true }
 insta = { workspace = true }
 insta-cmd = { workspace = true }
 markdown = { workspace = true }
+prek-consts = { workspace = true, features = ["test-utils"] }
 predicates = { workspace = true }
 pretty_assertions = { workspace = true }
 tempfile = { workspace = true }

crates/prek/src/cli/auto_update/mod.rs

@@ -15,7 +15,7 @@ use crate::cli::run::Selectors;
 use crate::config::GlobPatterns;
 use crate::fs::CWD;
 use crate::printer::Printer;
-use crate::run::CONCURRENCY;
+use crate::run::INTERNAL_CONCURRENCY;
 use crate::settings::FilesystemOptions;
 use crate::store::Store;
 use crate::workspace::{Project, Workspace};
@@ -393,7 +393,11 @@ pub(crate) async fn auto_update(
 
     let tag_filters =
         TagFilters::new(include_tag, exclude_tag, repo_include_tag, repo_exclude_tag)?;
-    let jobs = if jobs == 0 { *CONCURRENCY } else { jobs };
+    let jobs = if jobs == 0 {
+        *INTERNAL_CONCURRENCY
+    } else {
+        jobs
+    };
     let reporter = AutoUpdateReporter::new(printer);
 
     let repo_sources = collect_repo_sources(&workspace, cooldown_days, filesystem.as_ref())?;

crates/prek/src/cli/run/install.rs

@@ -11,7 +11,7 @@ use tracing::{debug, warn};
 use crate::cli::reporter::HookInstallReporter;
 use crate::config::Language;
 use crate::hook::{Hook, InstallInfo, InstalledHook};
-use crate::run::CONCURRENCY;
+use crate::run::INTERNAL_CONCURRENCY;
 use crate::store::Store;
 
 /// Resolve already-installed hook environments and install the missing ones.
@@ -41,7 +41,7 @@ pub(crate) async fn install_hooks(
         }
     }
 
-    let semaphore = Rc::new(Semaphore::new(*CONCURRENCY));
+    let semaphore = Rc::new(Semaphore::new(*INTERNAL_CONCURRENCY));
     let mut futures = FuturesUnordered::new();
 
     for partition in partition_hooks(hooks_to_install) {
@@ -284,7 +284,7 @@ impl InstallCache {
                         };
                         Some(CachedInstallInfo::new(Arc::new(info)))
                     })
-                    .buffer_unordered(*CONCURRENCY);
+                    .buffer_unordered(*INTERNAL_CONCURRENCY);
 
                 let mut hooks = Vec::new();
                 while let Some(hook) = tasks.next().await {

crates/prek/src/cli/run/run.rs

@@ -31,7 +31,7 @@ use crate::fs::CWD;
 use crate::git::GIT_ROOT;
 use crate::hook::{Hook, InstalledHook};
 use crate::printer::Printer;
-use crate::run::{CONCURRENCY, USE_COLOR};
+use crate::run::{HOOK_CONCURRENCY, USE_COLOR};
 use crate::store::Store;
 use crate::workspace::{HookInitFilters, Project, Workspace};
 use crate::{fs, git, hooks, warn_user};
@@ -729,7 +729,7 @@ impl<'a> HookRunSession<'a> {
         tag_cache: &FileTagCache<'paths>,
         clean_baseline: bool,
     ) -> Result<Vec<ProjectRunResult<'project, 'paths>>> {
-        let semaphore = Rc::new(Semaphore::new(*CONCURRENCY));
+        let semaphore = Rc::new(Semaphore::new(*HOOK_CONCURRENCY));
         let mut runs = FuturesUnordered::new();
         for (idx, project_run) in project_runs.into_iter().enumerate() {
             let semaphore = Rc::clone(&semaphore);
@@ -846,9 +846,8 @@ impl<'a> HookRunSession<'a> {
         semaphore: Rc<Semaphore>,
     ) -> Result<Vec<RunResult>> {
         debug!(
-            "Running priority group with priority {} with concurrency {}: {:?}",
+            "Running priority group with priority {}: {:?}",
             group_hooks[0].priority,
-            *CONCURRENCY,
             group_hooks.iter().map(|hook| &hook.id).collect::<Vec<_>>()
         );
 

crates/prek/src/hooks/builtin_hooks/check_json5.rs

@@ -3,15 +3,17 @@ use std::path::Path;
 use crate::hook::Hook;
 use crate::hooks::pre_commit_hooks::check_json::JsonDuplicateKeyChecker;
 use crate::hooks::run_concurrent_file_checks;
-use crate::run::CONCURRENCY;
+use crate::run::INTERNAL_CONCURRENCY;
 
 pub(crate) async fn check_json5(
     hook: &Hook,
     filenames: &[&Path],
 ) -> anyhow::Result<(i32, Vec<u8>)> {
-    run_concurrent_file_checks(filenames.iter().copied(), *CONCURRENCY, |filename| {
-        check_file(hook.project().relative_path(), filename)
-    })
+    run_concurrent_file_checks(
+        filenames.iter().copied(),
+        *INTERNAL_CONCURRENCY,
+        |filename| check_file(hook.project().relative_path(), filename),
+    )
     .await
 }
 

crates/prek/src/hooks/pre_commit_hooks/check_added_large_files.rs

@@ -6,7 +6,7 @@ use rustc_hash::FxHashSet;
 use crate::git::{get_added_files, get_lfs_files};
 use crate::hook::Hook;
 use crate::hooks::run_concurrent_file_checks;
-use crate::run::CONCURRENCY;
+use crate::run::INTERNAL_CONCURRENCY;
 
 #[derive(Parser)]
 #[command(disable_help_subcommand = true)]
@@ -54,7 +54,7 @@ pub(crate) async fn check_added_large_files(
         .copied()
         .filter(|f| !lfs_files.contains(*f));
 
-    run_concurrent_file_checks(filenames, *CONCURRENCY, |filename| async move {
+    run_concurrent_file_checks(filenames, *INTERNAL_CONCURRENCY, |filename| async move {
         let file_path = hook.project().relative_path().join(filename);
         let size = fs_err::tokio::metadata(file_path).await?.len() / 1024;
         if size > args.max_kb {

crates/prek/src/hooks/pre_commit_hooks/check_executables_have_shebangs.rs

@@ -9,7 +9,7 @@ use crate::hooks::pre_commit_hooks::shebangs::{
     file_has_shebang, git_index_stage_output, matching_git_index_paths_by_executable_bit,
 };
 use crate::hooks::run_concurrent_file_checks;
-use crate::run::CONCURRENCY;
+use crate::run::INTERNAL_CONCURRENCY;
 use rustc_hash::FxHashSet;
 
 pub(crate) async fn check_executables_have_shebangs(
@@ -47,16 +47,20 @@ async fn os_check_shebangs(
     file_base: &Path,
     paths: &[&Path],
 ) -> Result<(i32, Vec<u8>), anyhow::Error> {
-    run_concurrent_file_checks(paths.iter().copied(), *CONCURRENCY, |file| async move {
-        let file_path = file_base.join(file);
-        let has_shebang = file_has_shebang(&file_path).await?;
-        if has_shebang {
-            anyhow::Ok((0, Vec::new()))
-        } else {
-            let msg = build_missing_shebang_warning(file)?;
-            Ok((1, msg.into_bytes()))
-        }
-    })
+    run_concurrent_file_checks(
+        paths.iter().copied(),
+        *INTERNAL_CONCURRENCY,
+        |file| async move {
+            let file_path = file_base.join(file);
+            let has_shebang = file_has_shebang(&file_path).await?;
+            if has_shebang {
+                anyhow::Ok((0, Vec::new()))
+            } else {
+                let msg = build_missing_shebang_warning(file)?;
+                Ok((1, msg.into_bytes()))
+            }
+        },
+    )
     .await
 }
 
@@ -98,7 +102,7 @@ async fn git_check_shebangs(
     let filenames: FxHashSet<_> = filenames.iter().copied().collect();
     let entries = matching_git_index_paths_by_executable_bit(&stdout, file_base, &filenames, true);
 
-    run_concurrent_file_checks(entries, *CONCURRENCY, |file| async move {
+    run_concurrent_file_checks(entries, *INTERNAL_CONCURRENCY, |file| async move {
         let file_path = file_base.join(file);
         if file_has_shebang(&file_path).await? {
             Ok((0, Vec::new()))

crates/prek/src/hooks/pre_commit_hooks/check_json.rs

@@ -7,12 +7,14 @@ use serde::{Deserialize, Deserializer};
 
 use crate::hook::Hook;
 use crate::hooks::run_concurrent_file_checks;
-use crate::run::CONCURRENCY;
+use crate::run::INTERNAL_CONCURRENCY;
 
 pub(crate) async fn check_json(hook: &Hook, filenames: &[&Path]) -> Result<(i32, Vec<u8>)> {
-    run_concurrent_file_checks(filenames.iter().copied(), *CONCURRENCY, |filename| {
-        check_file(hook.project().relative_path(), filename)
-    })
+    run_concurrent_file_checks(
+        filenames.iter().copied(),
+        *INTERNAL_CONCURRENCY,
+        |filename| check_file(hook.project().relative_path(), filename),
+    )
     .await
 }
 

crates/prek/src/hooks/pre_commit_hooks/check_merge_conflict.rs

@@ -8,7 +8,7 @@ use tokio::io::AsyncBufReadExt;
 use crate::git::get_git_dir;
 use crate::hook::Hook;
 use crate::hooks::run_concurrent_file_checks;
-use crate::run::CONCURRENCY;
+use crate::run::INTERNAL_CONCURRENCY;
 
 const START_PATTERN: &[u8] = b"<<<<<<< ";
 const ANCESTOR_PATTERN: &[u8] = b"||||||| ";
@@ -35,9 +35,11 @@ pub(crate) async fn check_merge_conflict(
         return Ok((0, Vec::new()));
     }
 
-    run_concurrent_file_checks(filenames.iter().copied(), *CONCURRENCY, |filename| {
-        check_file(hook.project().relative_path(), filename)
-    })
+    run_concurrent_file_checks(
+        filenames.iter().copied(),
+        *INTERNAL_CONCURRENCY,
+        |filename| check_file(hook.project().relative_path(), filename),
+    )
     .await
 }
 

crates/prek/src/hooks/pre_commit_hooks/check_shebang_scripts_are_executable.rs

@@ -8,7 +8,7 @@ use crate::hooks::pre_commit_hooks::shebangs::{
     file_has_shebang, git_index_stage_output, matching_git_index_paths_by_executable_bit,
 };
 use crate::hooks::run_concurrent_file_checks;
-use crate::run::CONCURRENCY;
+use crate::run::INTERNAL_CONCURRENCY;
 use rustc_hash::FxHashSet;
 
 pub(crate) async fn check_shebang_scripts_are_executable(
@@ -24,7 +24,7 @@ pub(crate) async fn check_shebang_scripts_are_executable(
     let filenames: FxHashSet<_> = filenames.iter().copied().collect();
     let entries = matching_git_index_paths_by_executable_bit(&stdout, file_base, &filenames, false);
 
-    run_concurrent_file_checks(entries, *CONCURRENCY, |file| async move {
+    run_concurrent_file_checks(entries, *INTERNAL_CONCURRENCY, |file| async move {
         let file_path = file_base.join(file);
         if file_has_shebang(&file_path).await? {
             Ok((1, build_non_executable_shebang_warning(file)?.into_bytes()))

crates/prek/src/hooks/pre_commit_hooks/check_symlinks.rs

@@ -4,12 +4,14 @@ use anyhow::Result;
 
 use crate::hook::Hook;
 use crate::hooks::run_concurrent_file_checks;
-use crate::run::CONCURRENCY;
+use crate::run::INTERNAL_CONCURRENCY;
 
 pub(crate) async fn check_symlinks(hook: &Hook, filenames: &[&Path]) -> Result<(i32, Vec<u8>)> {
-    run_concurrent_file_checks(filenames.iter().copied(), *CONCURRENCY, |filename| {
-        check_file(hook.project().relative_path(), filename)
-    })
+    run_concurrent_file_checks(
+        filenames.iter().copied(),
+        *INTERNAL_CONCURRENCY,
+        |filename| check_file(hook.project().relative_path(), filename),
+    )
     .await
 }
 

crates/prek/src/hooks/pre_commit_hooks/check_toml.rs

@@ -5,12 +5,14 @@ use anyhow::Result;
 
 use crate::hook::Hook;
 use crate::hooks::run_concurrent_file_checks;
-use crate::run::CONCURRENCY;
+use crate::run::INTERNAL_CONCURRENCY;
 
 pub(crate) async fn check_toml(hook: &Hook, filenames: &[&Path]) -> Result<(i32, Vec<u8>)> {
-    run_concurrent_file_checks(filenames.iter().copied(), *CONCURRENCY, |filename| {
-        check_file(hook.project().relative_path(), filename)
-    })
+    run_concurrent_file_checks(
+        filenames.iter().copied(),
+        *INTERNAL_CONCURRENCY,
+        |filename| check_file(hook.project().relative_path(), filename),
+    )
     .await
 }