fix: dry run missing tasks with no script (#6473)

### Description

#6466 broke
`test/edit-turbo-json/task.t` as it moved the dry run execution to
happen after we early exit when commands don't exist.

This PR moves the branching to happen before the early exit and avoids
performing some of the work that's necessary for a real run, but
unneeded for dry run. I don't love the level of nesting in the `visit`
function, but it'll do for now.

### Testing Instructions

`tests/edit-turbo-json/task.t` now passes on the Rust codepath.


Closes TURBO-1682

---------

Co-authored-by: Chris Olszewski <Chris Olszewski>

crates/turborepo-cache/src/async_cache.rs

@@ -204,7 +204,7 @@ mod tests {
             token: "my-token".to_string(),
             team_slug: None,
         });
-        let mut async_cache = AsyncCache::new(&opts, &repo_root_path, api_client, api_auth, None)?;
+        let async_cache = AsyncCache::new(&opts, &repo_root_path, api_client, api_auth, None)?;
 
         // Ensure that the cache is empty
         let response = async_cache.exists(&hash).await;
@@ -279,7 +279,7 @@ mod tests {
             token: "my-token".to_string(),
             team_slug: None,
         });
-        let mut async_cache = AsyncCache::new(&opts, &repo_root_path, api_client, api_auth, None)?;
+        let async_cache = AsyncCache::new(&opts, &repo_root_path, api_client, api_auth, None)?;
 
         // Ensure that the cache is empty
         let response = async_cache.exists(&hash).await;
@@ -360,7 +360,7 @@ mod tests {
             token: "my-token".to_string(),
             team_slug: None,
         });
-        let mut async_cache = AsyncCache::new(&opts, &repo_root_path, api_client, api_auth, None)?;
+        let async_cache = AsyncCache::new(&opts, &repo_root_path, api_client, api_auth, None)?;
 
         // Ensure that the cache is empty
         let response = async_cache.exists(&hash).await;

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

@@ -165,7 +165,7 @@ impl TaskCache {
         Ok(log_writer)
     }
 
-    pub async fn exists(&mut self) -> Result<Option<CacheHitMetadata>, CacheError> {
+    pub async fn exists(&self) -> Result<Option<CacheHitMetadata>, CacheError> {
         self.run_cache.cache.exists(&self.hash).await
     }
 

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

@@ -211,45 +211,54 @@ impl<'a> Visitor<'a> {
                 info.clone(),
                 &task_hash,
             );
-            // TODO(gsoltis): if/when we fix https://github.com/vercel/turbo/issues/937
-            // the following block should never get hit. In the meantime, keep it after
-            // hashing so that downstream tasks can count on the hash existing
-            //
-            // bail if the script doesn't exist
-            let Some(_command) = command else { continue };
 
-            let workspace_directory = self.repo_root.resolve(workspace_info.package_path());
+            // here is where we do the logic split
+            match self.dry {
+                true => {
+                    let dry_run_exec_context =
+                        factory.dry_run_exec_context(info.clone(), task_cache);
+                    let tracker = self.run_tracker.track_task(info.into_owned());
+                    tasks.push(tokio::spawn(async move {
+                        dry_run_exec_context.execute_dry_run(tracker).await;
+                    }));
+                }
+                false => {
+                    // TODO(gsoltis): if/when we fix https://github.com/vercel/turbo/issues/937
+                    // the following block should never get hit. In the meantime, keep it after
+                    // hashing so that downstream tasks can count on the hash existing
+                    //
+                    // bail if the script doesn't exist
+                    if command.is_none() {
+                        continue;
+                    }
 
-            let mut exec_context = factory.exec_context(
-                info.clone(),
-                task_hash,
-                task_cache,
-                workspace_directory,
-                execution_env,
-            );
+                    let workspace_directory = self.repo_root.resolve(workspace_info.package_path());
 
-            let tracker = self.run_tracker.track_task(info.clone().into_owned());
-            let spaces_client = self.run_tracker.spaces_task_client();
-            let output_client = self.output_client();
-            let parent_span = Span::current();
-            let is_dry_run = self.dry;
-
-            if is_dry_run {
-                tasks.push(tokio::spawn(async move {
-                    exec_context.execute_dry_run(tracker).await;
-                }));
-            } else {
-                tasks.push(tokio::spawn(async move {
-                    exec_context
-                        .execute(
-                            parent_span.id(),
-                            tracker,
-                            output_client,
-                            callback,
-                            spaces_client,
-                        )
-                        .await;
-                }));
+                    let mut exec_context = factory.exec_context(
+                        info.clone(),
+                        task_hash,
+                        task_cache,
+                        workspace_directory,
+                        execution_env,
+                    );
+
+                    let tracker = self.run_tracker.track_task(info.clone().into_owned());
+                    let spaces_client = self.run_tracker.spaces_task_client();
+                    let output_client = self.output_client();
+                    let parent_span = Span::current();
+
+                    tasks.push(tokio::spawn(async move {
+                        exec_context
+                            .execute(
+                                parent_span.id(),
+                                tracker,
+                                output_client,
+                                callback,
+                                spaces_client,
+                            )
+                            .await;
+                    }));
+                }
             }
         }
 
@@ -552,6 +561,18 @@ impl<'a> ExecContextFactory<'a> {
             errors: self.errors.clone(),
         }
     }
+
+    pub fn dry_run_exec_context(
+        &self,
+        task_id: TaskId<'static>,
+        task_cache: TaskCache,
+    ) -> DryRunExecContext {
+        DryRunExecContext {
+            task_id,
+            task_cache,
+            hash_tracker: self.visitor.task_hasher.task_hash_tracker(),
+        }
+    }
 }
 
 struct ExecContext {
@@ -851,3 +872,20 @@ impl ExecContext {
         }
     }
 }
+
+struct DryRunExecContext {
+    task_id: TaskId<'static>,
+    task_cache: TaskCache,
+    hash_tracker: TaskHashTracker,
+}
+
+impl DryRunExecContext {
+    pub async fn execute_dry_run(&self, tracker: TaskTracker<()>) {
+        // may also need to do framework & command stuff?
+        if let Ok(Some(status)) = self.task_cache.exists().await {
+            self.hash_tracker
+                .insert_cache_status(self.task_id.clone(), status);
+        }
+        tracker.dry_run().await;
+    }
+}