perf: Early engine callback for confirmed cache hits

During the precompute phase, check local cache existence for each
task (a cheap stat call). For confirmed cache hits, send the engine
callback immediately in the dispatch loop instead of waiting for
the full cache restore cycle. This unblocks dependent tasks sooner,
reducing the DAG walk idle wait.

Exposes sync exists_local() through the cache stack so rayon
threads can check cache existence without async overhead.

Author: anthonyshew

crates/turborepo-cache/src/async_cache.rs

@@ -165,6 +165,11 @@ impl AsyncCache {
         self.real_cache.exists(key).await
     }
 
+    /// Synchronous local-only cache existence check.
+    pub fn exists_local(&self, key: &str) -> Result<Option<CacheHitMetadata>, CacheError> {
+        self.real_cache.exists_local(key)
+    }
+
     #[tracing::instrument(skip_all)]
     pub async fn fetch(
         &self,

crates/turborepo-cache/src/fs.rs

@@ -124,7 +124,7 @@ impl FSCache {
     }
 
     #[tracing::instrument(skip_all)]
-    pub(crate) fn exists(&self, hash: &str) -> Result<Option<CacheHitMetadata>, CacheError> {
+    pub fn exists(&self, hash: &str) -> Result<Option<CacheHitMetadata>, CacheError> {
         let cache_dir = self.cache_directory.as_str();
         let mut buf = String::with_capacity(cache_dir.len() + 1 + hash.len() + "-meta.json".len());
         buf.push_str(cache_dir);

crates/turborepo-cache/src/multiplexer.rs

@@ -180,30 +180,39 @@ impl CacheMultiplexer {
 
     #[tracing::instrument(skip_all)]
     pub async fn exists(&self, key: &str) -> Result<Option<CacheHitMetadata>, CacheError> {
-        if self.cache_config.local.read
-            && let Some(fs) = &self.fs
+        if let Some(hit) = self.exists_local(key)? {
+            return Ok(Some(hit));
+        }
+
+        if self.cache_config.remote.read
+            && let Some(http) = self.get_http_cache()
         {
-            match fs.exists(key) {
+            match http.exists(key).await {
                 cache_hit @ Ok(Some(_)) => {
                     return cache_hit;
                 }
                 Ok(None) => {}
-                Err(err) => debug!("failed to check fs cache: {:?}", err),
+                Err(err) => debug!("failed to check http cache: {:?}", err),
             }
         }
 
-        if self.cache_config.remote.read
-            && let Some(http) = self.get_http_cache()
+        Ok(None)
+    }
+
+    /// Synchronous local-only cache existence check. Safe to call from
+    /// rayon threads or any non-async context.
+    pub fn exists_local(&self, key: &str) -> Result<Option<CacheHitMetadata>, CacheError> {
+        if self.cache_config.local.read
+            && let Some(fs) = &self.fs
         {
-            match http.exists(key).await {
+            match fs.exists(key) {
                 cache_hit @ Ok(Some(_)) => {
                     return cache_hit;
                 }
                 Ok(None) => {}
-                Err(err) => debug!("failed to check http cache: {:?}", err),
+                Err(err) => debug!("failed to check fs cache: {:?}", err),
             }
         }
-
         Ok(None)
     }
 }

crates/turborepo-lib/src/task_graph/visitor/mod.rs

@@ -46,6 +46,15 @@ use crate::{
     },
 };
 
+/// Pre-computed data for a single task: hash, execution environment,
+/// and whether the local cache already contains a hit for this hash.
+#[derive(Debug)]
+struct PrecomputedTask {
+    hash: String,
+    execution_env: EnvironmentVariableMap,
+    cache_hit: bool,
+}
+
 // This holds the whole world
 pub struct Visitor<'a> {
     color_cache: ColorSelector,
@@ -186,13 +195,13 @@ impl<'a> Visitor<'a> {
 
     /// Pre-compute task hashes and execution environments for all tasks in
     /// parallel. Tasks are processed in topological waves so dependency
-    /// hashes are always available when needed. Returns a map from TaskId
-    /// to (hash, execution_env).
+    /// hashes are always available when needed. Also checks cache existence
+    /// for each task so the dispatch loop can fast-path cache hits.
     fn precompute_task_hashes(
         &self,
         engine: &Engine,
         telemetry: &GenericEventBuilder,
-    ) -> Result<HashMap<TaskId<'static>, (String, EnvironmentVariableMap)>, Error> {
+    ) -> Result<HashMap<TaskId<'static>, PrecomputedTask>, Error> {
         use petgraph::algo::toposort;
         use rayon::prelude::*;
         use turborepo_engine::TaskNode;
@@ -230,14 +239,13 @@ impl<'a> Visitor<'a> {
             waves[d].push(node_idx);
         }
 
-        let results: Arc<Mutex<HashMap<TaskId<'static>, (String, EnvironmentVariableMap)>>> =
+        let results: Arc<Mutex<HashMap<TaskId<'static>, PrecomputedTask>>> =
             Arc::new(Mutex::new(HashMap::with_capacity(sorted.len())));
 
         // Process each wave in parallel. Within a wave, all dependencies
         // have already been hashed in earlier waves.
         for wave in &waves {
-            type HashResult =
-                Result<Option<(TaskId<'static>, String, EnvironmentVariableMap)>, Error>;
+            type HashResult = Result<Option<(TaskId<'static>, PrecomputedTask)>, Error>;
             let wave_results: Vec<HashResult> = wave
                 .par_iter()
                 .map(|&node_idx| {
@@ -284,14 +292,24 @@ impl<'a> Visitor<'a> {
                         self.task_hasher
                             .env(task_id, task_env_mode, task_definition)?;
 
-                    Ok(Some((task_id.clone(), task_hash, execution_env)))
+                    let cache_hit =
+                        task_definition.cache && self.run_cache.exists_local(&task_hash);
+
+                    Ok(Some((
+                        task_id.clone(),
+                        PrecomputedTask {
+                            hash: task_hash,
+                            execution_env,
+                            cache_hit,
+                        },
+                    )))
                 })
                 .collect();
 
             let mut map = results.lock().expect("precompute lock poisoned");
             for result in wave_results {
-                if let Some((task_id, hash, env)) = result? {
-                    map.insert(task_id, (hash, env));
+                if let Some((task_id, precomputed)) = result? {
+                    map.insert(task_id, precomputed);
                 }
             }
         }
@@ -381,11 +399,11 @@ impl<'a> Visitor<'a> {
                 .task_definition(&info)
                 .ok_or(Error::MissingDefinition)?;
 
-            // Look up pre-computed hash and env instead of computing them here.
-            let (task_hash, execution_env) = precomputed
-                .get(&info)
-                .ok_or(Error::MissingDefinition)?
-                .clone();
+            // Look up pre-computed hash, env, and cache status.
+            let precomputed_task = precomputed.get(&info).ok_or(Error::MissingDefinition)?;
+            let task_hash = precomputed_task.hash.clone();
+            let execution_env = precomputed_task.execution_env.clone();
+            let confirmed_cache_hit = precomputed_task.cache_hit;
 
             debug!("task {} hash is {}", info, task_hash);
 
@@ -395,6 +413,16 @@ impl<'a> Visitor<'a> {
                     .task_cache(task_definition, workspace_info, info.clone(), &task_hash)
             };
 
+            // For confirmed cache hits (checked in parallel during
+            // precompute), signal the engine immediately so dependents
+            // can start without waiting for the full cache restore.
+            let callback = if confirmed_cache_hit {
+                callback.send(Ok(())).ok();
+                None
+            } else {
+                Some(callback)
+            };
+
             // Drop to avoid holding the span across an await
 
             drop(_enter);

crates/turborepo-run-cache/src/lib.rs

@@ -102,6 +102,15 @@ impl RunCache {
         }
     }
 
+    /// Synchronous local-only cache existence check. Can be called from
+    /// rayon threads during hash precomputation.
+    pub fn exists_local(&self, hash: &str) -> bool {
+        if self.reads_disabled {
+            return false;
+        }
+        matches!(self.cache.exists_local(hash), Ok(Some(_)))
+    }
+
     pub fn task_cache(
         self: &Arc<Self>,
         // TODO: Group these in a struct

crates/turborepo-task-executor/src/exec.rs

@@ -265,7 +265,7 @@ where
         parent_span_id: Option<tracing::Id>,
         tracker: TaskTracker<()>,
         output_client: TaskOutput<O>,
-        callback: oneshot::Sender<Result<(), StopExecution>>,
+        callback: Option<oneshot::Sender<Result<(), StopExecution>>>,
         telemetry: &PackageTaskEventBuilder,
     ) -> Result<(), InternalError> {
         let tracker: TaskTracker<chrono::DateTime<chrono::Local>> = tracker.start().await;
@@ -293,17 +293,20 @@ where
                     SuccessOutcome::CacheHit => tracker.cached().await,
                     SuccessOutcome::Run => tracker.build_succeeded(0).await,
                 };
-                callback.send(Ok(())).ok();
+                if let Some(cb) = callback {
+                    cb.send(Ok(())).ok();
+                }
             }
             Ok(ExecOutcome::Task { exit_code, message }) => {
                 tracker.build_failed(exit_code, message).await;
-                callback
-                    .send(match self.continue_on_error {
+                if let Some(cb) = callback {
+                    cb.send(match self.continue_on_error {
                         ContinueMode::Always => Ok(()),
                         ContinueMode::DependenciesSuccessful => Err(StopExecution::DependentTasks),
                         ContinueMode::Never => Err(StopExecution::AllTasks),
                     })
                     .ok();
+                }
 
                 match self.continue_on_error {
                     ContinueMode::Always | ContinueMode::DependenciesSuccessful => (),
@@ -312,16 +315,22 @@ where
             }
             Ok(ExecOutcome::Shutdown) => {
                 tracker.cancel();
-                callback.send(Err(StopExecution::AllTasks)).ok();
+                if let Some(cb) = callback {
+                    cb.send(Err(StopExecution::AllTasks)).ok();
+                }
                 self.manager.stop().await;
             }
             Ok(ExecOutcome::Restarted) => {
                 tracker.cancel();
-                callback.send(Err(StopExecution::DependentTasks)).ok();
+                if let Some(cb) = callback {
+                    cb.send(Err(StopExecution::DependentTasks)).ok();
+                }
             }
             Err(e) => {
                 tracker.cancel();
-                callback.send(Err(StopExecution::AllTasks)).ok();
+                if let Some(cb) = callback {
+                    cb.send(Err(StopExecution::AllTasks)).ok();
+                }
                 self.manager.stop().await;
                 return Err(e);
             }