From eaf7897638d58660cc9aa7d227e5666b08fd6ab8 Mon Sep 17 00:00:00 2001 From: Anthony Shew Date: Mon, 15 Jun 2026 10:59:59 -0600 Subject: [PATCH] fix: Precompute unblocked JIT descendants --- .../src/task_graph/visitor/mod.rs | 169 ++++++++++++++---- crates/turborepo/ARCHITECTURE.md | 2 + crates/turborepo/tests/run_caching.rs | 66 +++++++ 3 files changed, 198 insertions(+), 39 deletions(-) diff --git a/crates/turborepo-lib/src/task_graph/visitor/mod.rs b/crates/turborepo-lib/src/task_graph/visitor/mod.rs index 052f3f7fb83d8..18b8d811674b7 100644 --- a/crates/turborepo-lib/src/task_graph/visitor/mod.rs +++ b/crates/turborepo-lib/src/task_graph/visitor/mod.rs @@ -35,7 +35,7 @@ use turborepo_types::{EnvMode, ResolvedLogOrder, ResolvedLogPrefix}; use turborepo_ui::{sender::UISender, ColorConfig, ColorSelector}; use crate::{ - engine::{Engine, ExecutionOptions}, + engine::{Engine, ExecutionOptions, TaskNode}, microfrontends::MicrofrontendsConfigs, opts::RunOpts, run::{task_access::TaskAccess, RunCache}, @@ -216,6 +216,125 @@ impl<'a> Visitor<'a> { /// parallel. Tasks are processed in topological waves so dependency /// hashes are always available when needed. Returns a map from TaskId /// to (hash, execution_env). + fn precompute_ready_task_hash( + &self, + engine: &Engine, + telemetry: &GenericEventBuilder, + task_id: &TaskId<'static>, + ) -> Result { + let package_name = PackageName::from(task_id.package()); + let workspace_info = self + .package_graph + .package_info(&package_name) + .ok_or_else(|| Error::MissingPackage { + package_name: package_name.clone(), + task_id: task_id.clone(), + })?; + + let task_definition = engine + .task_definition(task_id) + .ok_or(Error::MissingDefinition)?; + + let task_env_mode = task_definition.env_mode.unwrap_or(self.global_env_mode); + + let dependency_set = engine + .dependencies(task_id) + .ok_or(Error::MissingDefinition)?; + + let package_task_event = + PackageTaskEventBuilder::new(task_id.package(), task_id.task()).with_parent(telemetry); + package_task_event.track_env_mode(&task_env_mode.to_string()); + + let task_hash_telemetry = package_task_event.child(); + let task_hash = self.task_hasher.calculate_task_hash( + task_id, + task_definition, + task_env_mode, + workspace_info, + &dependency_set, + task_hash_telemetry, + )?; + + let execution_env = self + .task_hasher + .env(task_id, task_env_mode, task_definition)?; + + Ok(PrecomputedTask::Ready { + task_hash, + execution_env, + }) + } + + fn dependency_hashes_available( + &self, + engine: &Engine, + task_id: &TaskId<'static>, + ) -> Result { + let dependency_set = engine + .dependencies(task_id) + .ok_or(Error::MissingDefinition)?; + let task_hash_tracker = self.task_hasher.task_hash_tracker(); + + Ok(dependency_set.iter().all(|dependency| { + let TaskNode::Task(dependency_task_id) = dependency else { + return true; + }; + + task_hash_tracker.hash(dependency_task_id).is_some() + })) + } + + fn precompute_unblocked_deferred_hashes( + &self, + engine: &Engine, + telemetry: &GenericEventBuilder, + precomputed: &mut HashMap, PrecomputedTask>, + ) -> Result<(), Error> { + use rayon::prelude::*; + + loop { + let ready_to_hash = precomputed + .iter() + .filter_map(|(task_id, precomputed_task)| { + if !matches!(precomputed_task, PrecomputedTask::Deferred) { + return None; + } + + let Some(task_definition) = engine.task_definition(task_id) else { + return Some(Err(Error::MissingDefinition)); + }; + if task_definition.inputs.has_jit_inputs() { + return None; + } + + match self.dependency_hashes_available(engine, task_id) { + Ok(true) => Some(Ok(task_id.clone())), + Ok(false) => None, + Err(err) => Some(Err(err)), + } + }) + .collect::, _>>()?; + + if ready_to_hash.is_empty() { + return Ok(()); + } + + type HashResult = Result<(TaskId<'static>, PrecomputedTask), Error>; + let hash_results: Vec = ready_to_hash + .par_iter() + .map(|task_id| { + self.precompute_ready_task_hash(engine, telemetry, task_id) + .map(|precomputed_task| (task_id.clone(), precomputed_task)) + }) + .collect(); + + for result in hash_results { + let (task_id, precomputed_task) = result?; + precomputed.insert(task_id, precomputed_task); + } + } + } + fn precompute_task_hashes( &self, engine: &Engine, @@ -223,8 +342,6 @@ impl<'a> Visitor<'a> { ) -> Result, PrecomputedTask>, Error> { use petgraph::algo::toposort; use rayon::prelude::*; - use turborepo_engine::TaskNode; - let graph = engine.task_graph(); let mut sorted = toposort(graph, None).map_err(|_| Error::MissingDefinition)?; // toposort returns dependents before dependencies (edges point @@ -273,19 +390,9 @@ impl<'a> Visitor<'a> { return Ok(None); }; - let package_name = PackageName::from(task_id.package()); - let workspace_info = self - .package_graph - .package_info(&package_name) - .ok_or_else(|| Error::MissingPackage { - package_name: package_name.clone(), - task_id: task_id.clone(), - })?; - let task_definition = engine .task_definition(task_id) .ok_or(Error::MissingDefinition)?; - let task_env_mode = task_definition.env_mode.unwrap_or(self.global_env_mode); let dependency_set = engine @@ -316,32 +423,8 @@ impl<'a> Visitor<'a> { return Ok(Some((task_id.clone(), PrecomputedTask::Deferred))); } - let package_task_event = - PackageTaskEventBuilder::new(task_id.package(), task_id.task()) - .with_parent(telemetry); - package_task_event.track_env_mode(&task_env_mode.to_string()); - - let task_hash_telemetry = package_task_event.child(); - let task_hash = self.task_hasher.calculate_task_hash( - task_id, - task_definition, - task_env_mode, - workspace_info, - &dependency_set, - task_hash_telemetry, - )?; - - let execution_env = - self.task_hasher - .env(task_id, task_env_mode, task_definition)?; - - Ok(Some(( - task_id.clone(), - PrecomputedTask::Ready { - task_hash, - execution_env, - }, - ))) + self.precompute_ready_task_hash(engine, telemetry, task_id) + .map(|precomputed_task| Some((task_id.clone(), precomputed_task))) }) .collect(); @@ -528,6 +611,14 @@ impl<'a> Visitor<'a> { break; } }; + if let Err(err) = self.precompute_unblocked_deferred_hashes( + &engine, + telemetry, + &mut precomputed, + ) { + dispatch_error = Some(err); + break; + } (task_hash, execution_env) } } diff --git a/crates/turborepo/ARCHITECTURE.md b/crates/turborepo/ARCHITECTURE.md index c706e2949d0cd..9040d3138e80c 100644 --- a/crates/turborepo/ARCHITECTURE.md +++ b/crates/turborepo/ARCHITECTURE.md @@ -194,6 +194,8 @@ The task graph visitor handles task execution: engine dispatches the task, after its dependencies have completed and restored any cached outputs. Tasks that depend on deferred tasks are also deferred so their dependency hashes are available before their own hash is calculated. + Once a deferred task has a real hash, the visitor precomputes any unblocked + non-JIT descendants instead of waiting for each descendant to be dispatched. - Creates `ExecContext` for each task - Manages UI output and progress tracking - Collects errors and execution information diff --git a/crates/turborepo/tests/run_caching.rs b/crates/turborepo/tests/run_caching.rs index 26716ee37523f..112761e402d19 100644 --- a/crates/turborepo/tests/run_caching.rs +++ b/crates/turborepo/tests/run_caching.rs @@ -367,6 +367,72 @@ fn test_jit_dependency_defers_dependent_hashing() { ); } +#[test] +fn test_jit_descendant_hashes_after_jit_hash_is_available() { + let tempdir = tempfile::tempdir().unwrap(); + setup::setup_integration_test(tempdir.path(), "basic_monorepo", "npm@10.5.0", true).unwrap(); + + fs::write(tempdir.path().join("apps/my-app/marker.txt"), "before\n").unwrap(); + fs::write(tempdir.path().join("apps/my-app/jit-input.txt"), "stable\n").unwrap(); + fs::write( + tempdir.path().join("apps/my-app/package.json"), + r#"{ + "name": "my-app", + "scripts": { + "generate": "node -e \"const fs = require('fs'); fs.mkdirSync('.generated', { recursive: true }); fs.writeFileSync('.generated/done.txt', 'done\\n'); fs.writeFileSync('marker.txt', 'after\\n')\"", + "build": "node -e \"const fs = require('fs'); fs.mkdirSync('.output', { recursive: true }); fs.writeFileSync('.output/marker.txt', fs.readFileSync('marker.txt'))\"" + }, + "dependencies": { + "util": "*" + } +} +"#, + ) + .unwrap(); + + fs::write( + tempdir.path().join("turbo.json"), + r#"{ + "$schema": "https://turborepo.dev/schema.json", + "tasks": { + "generate": { + "inputs": ["$TURBO_JIT$/jit-input.txt"], + "outputs": [".generated/**"] + }, + "build": { + "dependsOn": ["generate"], + "inputs": ["marker.txt"], + "outputs": [".output/**"] + } + } +} +"#, + ) + .unwrap(); + + let output = run_turbo( + tempdir.path(), + &["run", "build", "--filter=my-app", "--output-logs=none"], + ); + let stdout = String::from_utf8_lossy(&output.stdout); + assert!( + stdout.contains("0 cached, 2 total"), + "expected first run to execute both tasks, got:\n{stdout}" + ); + + fs::write(tempdir.path().join("apps/my-app/marker.txt"), "before\n").unwrap(); + + let output = run_turbo( + tempdir.path(), + &["run", "build", "--filter=my-app", "--output-logs=none"], + ); + let stdout = String::from_utf8_lossy(&output.stdout); + assert!( + stdout.contains("2 cached, 2 total"), + "expected JIT descendants to be hashed before the dependency command runs, got:\n{stdout}" + ); +} + #[test] fn test_gitignored_output_deletion_restores_from_cache() { let tempdir = tempfile::tempdir().unwrap();