fix(turborepo): Run outline fixes (#5966)

### Description
Found a lot of bugs while testing #5851, so split it off into a separate
PR.

### Testing Instructions

<!--
  Give a quick description of steps to test your changes.
-->


Closes TURBO-1334

---------

Co-authored-by: Alexander Lyon <Alexander Lyon>
Co-authored-by: nicholaslyang <Nicholas Yang>

crates/turborepo-env/src/lib.rs

@@ -14,6 +14,7 @@ use thiserror::Error;
 const DEFAULT_ENV_VARS: [&str; 1] = ["VERCEL_ANALYTICS_ID"];
 
 /// Environment mode after we've resolved the `Infer` variant
+#[derive(Debug)]
 pub enum ResolvedEnvMode {
     Loose,
     Strict,

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

@@ -4,7 +4,7 @@ use tracing::{debug, error};
 use crate::{commands::CommandBase, run::Run};
 
 pub async fn run(base: CommandBase) -> Result<()> {
-    let mut run = Run::new(base);
+    let mut run = Run::new(&base);
     debug!("using the experimental rust codepath");
     debug!("configured run struct: {:?}", run);
 

crates/turborepo-lib/src/config/turbo.rs

@@ -13,7 +13,7 @@ use crate::{
     config::Error,
     package_json::PackageJson,
     run::task_id::{TaskId, TaskName, ROOT_PKG_NAME},
-    task_graph::{BookkeepingTaskDefinition, Pipeline, TaskDefinitionHashable, TaskOutputs},
+    task_graph::{BookkeepingTaskDefinition, Pipeline, TaskDefinitionStable, TaskOutputs},
 };
 
 #[derive(Serialize, Deserialize, Debug, Default, PartialEq, Clone)]
@@ -31,7 +31,7 @@ pub struct TurboJson {
     pub(crate) global_deps: Vec<String>,
     pub(crate) global_dot_env: Vec<RelativeUnixPathBuf>,
     pub(crate) global_env: Vec<String>,
-    pub(crate) global_pass_through_env: Vec<String>,
+    pub(crate) global_pass_through_env: Option<Vec<String>>,
     pub(crate) pipeline: Pipeline,
     pub(crate) remote_cache_options: Option<RemoteCacheOpts>,
     pub(crate) space_id: Option<String>,
@@ -234,8 +234,7 @@ impl TryFrom<RawTaskDefinition> for BookkeepingTaskDefinition {
                 pass_through_env.sort();
                 Ok(pass_through_env)
             })
-            .transpose()?
-            .unwrap_or_default();
+            .transpose()?;
 
         let dot_env = raw_task
             .dot_env
@@ -265,7 +264,7 @@ impl TryFrom<RawTaskDefinition> for BookkeepingTaskDefinition {
             defined_fields,
             experimental_fields: Default::default(),
             experimental: Default::default(),
-            task_definition: TaskDefinitionHashable {
+            task_definition: TaskDefinitionStable {
                 outputs,
                 cache,
                 topological_dependencies,
@@ -351,8 +350,7 @@ impl TryFrom<RawTurboJSON> for TurboJson {
                     global_pass_through_env.sort();
                     Ok(global_pass_through_env)
                 })
-                .transpose()?
-                .unwrap_or_default(),
+                .transpose()?,
             global_deps: {
                 let mut global_deps: Vec<_> = global_file_dependencies.into_iter().collect();
                 global_deps.sort();
@@ -455,9 +453,9 @@ impl TurboJson {
                             set.insert("Cache".to_string());
                             set
                         },
-                        task_definition: TaskDefinitionHashable {
+                        task_definition: TaskDefinitionStable {
                             cache: false,
-                            ..TaskDefinitionHashable::default()
+                            ..TaskDefinitionStable::default()
                         },
                         ..BookkeepingTaskDefinition::default()
                     },
@@ -567,8 +565,7 @@ mod tests {
         package_json::PackageJson,
         run::task_id::TaskName,
         task_graph::{
-            BookkeepingTaskDefinition, TaskDefinitionExperiments, TaskDefinitionHashable,
-            TaskOutputs,
+            BookkeepingTaskDefinition, TaskDefinitionExperiments, TaskDefinitionStable, TaskOutputs,
         },
     };
 
@@ -587,7 +584,7 @@ mod tests {
     ; "global dot env (unsorted)")]
     #[test_case(r#"{ "globalPassThroughEnv": ["GITHUB_TOKEN", "AWS_SECRET_KEY"] }"#,
         TurboJson {
-            global_pass_through_env: vec!["AWS_SECRET_KEY".to_string(), "GITHUB_TOKEN".to_string()],
+            global_pass_through_env: Some(vec!["AWS_SECRET_KEY".to_string(), "GITHUB_TOKEN".to_string()]),
             ..TurboJson::default()
         }
     )]
@@ -617,9 +614,9 @@ mod tests {
                 "//#build".into(),
                 BookkeepingTaskDefinition {
                     defined_fields: ["Cache".to_string()].into_iter().collect(),
-                    task_definition: TaskDefinitionHashable {
+                    task_definition: TaskDefinitionStable {
                         cache: false,
-                        ..TaskDefinitionHashable::default()
+                        ..TaskDefinitionStable::default()
                     },
                     ..BookkeepingTaskDefinition::default()
                 }
@@ -652,9 +649,9 @@ mod tests {
                 "//#build".into(),
                 BookkeepingTaskDefinition {
                     defined_fields: ["Cache".to_string()].into_iter().collect(),
-                    task_definition: TaskDefinitionHashable {
+                    task_definition: TaskDefinitionStable {
                         cache: true,
-                        ..TaskDefinitionHashable::default()
+                        ..TaskDefinitionStable::default()
                     },
                     ..BookkeepingTaskDefinition::default()
                 }
@@ -663,9 +660,9 @@ mod tests {
                 "//#test".into(),
                 BookkeepingTaskDefinition {
                     defined_fields: ["Cache".to_string()].into_iter().collect(),
-                    task_definition: TaskDefinitionHashable {
+                    task_definition: TaskDefinitionStable {
                         cache: false,
-                        ..TaskDefinitionHashable::default()
+                        ..TaskDefinitionStable::default()
                     },
                     ..BookkeepingTaskDefinition::default()
                 }
@@ -706,7 +703,7 @@ mod tests {
             defined_fields: ["Persistent".to_string()].into_iter().collect(),
             experimental_fields: HashSet::new(),
             experimental: TaskDefinitionExperiments::default(),
-            task_definition: TaskDefinitionHashable::default()
+            task_definition: TaskDefinitionStable::default()
         }
     )]
     #[test_case(
@@ -746,7 +743,7 @@ mod tests {
             ].into_iter().collect(),
             experimental_fields: HashSet::new(),
             experimental: TaskDefinitionExperiments {},
-            task_definition: TaskDefinitionHashable {
+            task_definition: TaskDefinitionStable {
                 dot_env: vec![RelativeUnixPathBuf::new("package/a/.env").unwrap()],
                 env: vec!["OS".to_string()],
                 outputs: TaskOutputs {
@@ -756,7 +753,7 @@ mod tests {
                 cache: false,
                 inputs: vec!["package/a/src/**".to_string()],
                 output_mode: OutputLogsMode::Full,
-                pass_through_env: vec!["AWS_SECRET_KEY".to_string()],
+                pass_through_env: Some(vec!["AWS_SECRET_KEY".to_string()]),
                 task_dependencies: vec!["cli#build".into()],
                 topological_dependencies: vec![],
                 persistent: true,

crates/turborepo-lib/src/engine/builder.rs

@@ -6,6 +6,7 @@ use turbopath::AbsoluteSystemPath;
 use super::Engine;
 use crate::{
     config::{validate_extends, validate_no_package_task_syntax, TurboJson},
+    graph,
     package_graph::{PackageGraph, WorkspaceName, WorkspaceNode},
     run::task_id::{TaskId, TaskName, ROOT_PKG_NAME},
     task_graph::{BookkeepingTaskDefinition, TaskDefinition},
@@ -30,6 +31,8 @@ pub enum Error {
     Config(#[from] crate::config::Error),
     #[error("Invalid turbo.json:\n{error_lines}")]
     Validation { error_lines: String },
+    #[error(transparent)]
+    Graph(#[from] graph::Error),
 }
 
 pub struct EngineBuilder<'a> {
@@ -128,6 +131,13 @@ impl<'a> EngineBuilder<'a> {
         }
 
         if !missing_tasks.is_empty() {
+            let mut missing_tasks = missing_tasks
+                .into_iter()
+                .map(|task_name| task_name.to_string())
+                .collect::<Vec<_>>();
+            // We sort the tasks mostly to keep it deterministic for our tests
+            missing_tasks.sort();
+
             return Err(Error::MissingTasks(missing_tasks.into_iter().join(", ")));
         }
 
@@ -238,6 +248,8 @@ impl<'a> EngineBuilder<'a> {
             }
         }
 
+        graph::validate_graph(&engine.task_graph)?;
+
         Ok(engine.seal())
     }
 
@@ -868,9 +880,10 @@ mod test {
             WorkspaceName::Root,
             turbo_json(json!({
                 "pipeline": {
-                    "libA#build": { "dependsOn": ["app1#compile", "app1#build"] },
+                    "libA#build": { "dependsOn": ["app1#compile", "app1#test"] },
                     "build": { "dependsOn": ["^build"] },
-                    "compile": { "dependsOn": ["^build"] }
+                    "compile": {},
+                    "test": {}
                 }
             })),
         )]
@@ -889,9 +902,10 @@ mod test {
             .unwrap();
 
         let expected = deps! {
+            "app1#compile" => ["___ROOT___"],
+            "app1#test" => ["___ROOT___"],
             "app1#build" => ["libA#build"],
-            "app1#compile" => ["libA#build"],
-            "libA#build" => ["app1#build", "app1#compile"]
+            "libA#build" => ["app1#compile", "app1#test"]
         };
         assert_eq!(all_dependencies(&engine), expected);
     }

crates/turborepo-lib/src/graph/mod.rs

@@ -1,3 +1,44 @@
 mod walker;
 
+use std::fmt::Display;
+
+use itertools::Itertools;
+use petgraph::prelude::*;
+use thiserror::Error;
+
+#[derive(Debug, Error)]
+pub enum Error {
+    #[error("cyclic dependency detected:\n{0}")]
+    CyclicDependencies(String),
+    #[error("{0} depends on itself")]
+    SelfDependency(String),
+}
+
+pub fn validate_graph<G: Display>(graph: &Graph<G, ()>) -> Result<(), Error> {
+    // This is equivalent to AcyclicGraph.Cycles from Go's dag library
+    let cycles_lines = petgraph::algo::tarjan_scc(&graph)
+        .into_iter()
+        .filter(|cycle| cycle.len() > 1)
+        .map(|cycle| {
+            let workspaces = cycle.into_iter().map(|id| graph.node_weight(id).unwrap());
+            format!("\t{}", workspaces.format(", "))
+        })
+        .join("\n");
+
+    if !cycles_lines.is_empty() {
+        return Err(Error::CyclicDependencies(cycles_lines));
+    }
+
+    for edge in graph.edge_references() {
+        if edge.source() == edge.target() {
+            let node = graph
+                .node_weight(edge.source())
+                .expect("edge pointed to missing node");
+            return Err(Error::SelfDependency(node.to_string()));
+        }
+    }
+
+    Ok(())
+}
+
 pub use walker::{WalkMessage, Walker};

crates/turborepo-lib/src/hash/mod.rs

@@ -41,15 +41,16 @@ mod proto_capnp {
     }
 }
 
+#[derive(Debug)]
 pub struct TaskHashable<'a> {
     // hashes
     pub(crate) global_hash: &'a str,
     pub(crate) task_dependency_hashes: Vec<String>,
     pub(crate) hash_of_files: &'a str,
-    pub(crate) external_deps_hash: String,
+    pub(crate) external_deps_hash: Option<String>,
 
     // task
-    pub(crate) package_dir: turbopath::RelativeUnixPathBuf,
+    pub(crate) package_dir: Option<turbopath::RelativeUnixPathBuf>,
     pub(crate) task: &'a str,
     pub(crate) outputs: TaskOutputs,
     pub(crate) pass_through_args: &'a [String],
@@ -62,11 +63,12 @@ pub struct TaskHashable<'a> {
     pub(crate) dot_env: &'a [turbopath::RelativeUnixPathBuf],
 }
 
-#[derive(Debug)]
+#[derive(Debug, Clone)]
 pub struct GlobalHashable<'a> {
     pub global_cache_key: &'static str,
     pub global_file_hash_map: HashMap<turbopath::RelativeUnixPathBuf, String>,
-    pub root_external_dependencies_hash: String,
+    // This is None in single package mode
+    pub root_external_dependencies_hash: Option<String>,
     pub env: &'a [String],
     pub resolved_env_vars: Vec<String>,
     pub pass_through_env: &'a [String],
@@ -198,9 +200,15 @@ impl From<TaskHashable<'_>> for Builder<HeapAllocator> {
         let mut builder = message.init_root();
 
         builder.set_global_hash(task_hashable.global_hash);
-        builder.set_package_dir(&task_hashable.package_dir.to_string());
+        if let Some(package_dir) = task_hashable.package_dir {
+            builder.set_package_dir(&package_dir.to_string());
+        }
+
         builder.set_hash_of_files(task_hashable.hash_of_files);
-        builder.set_external_deps_hash(&task_hashable.external_deps_hash);
+        if let Some(external_deps_hash) = task_hashable.external_deps_hash {
+            builder.set_external_deps_hash(&external_deps_hash);
+        }
+
         builder.set_task(task_hashable.task);
         builder.set_env_mode(task_hashable.env_mode.into());
 
@@ -221,11 +229,11 @@ impl From<TaskHashable<'_>> for Builder<HeapAllocator> {
         }
 
         {
-            let mut pass_thru_args_builder = builder
+            let mut pass_through_args_builder = builder
                 .reborrow()
                 .init_pass_thru_args(task_hashable.pass_through_args.len() as u32);
             for (i, arg) in task_hashable.pass_through_args.iter().enumerate() {
-                pass_thru_args_builder.set(i as u32, arg);
+                pass_through_args_builder.set(i as u32, arg);
             }
         }
 
@@ -237,11 +245,11 @@ impl From<TaskHashable<'_>> for Builder<HeapAllocator> {
         }
 
         {
-            let mut pass_thru_env_builder = builder
+            let mut pass_through_env_builder = builder
                 .reborrow()
                 .init_pass_thru_env(task_hashable.pass_through_env.len() as u32);
             for (i, env) in task_hashable.pass_through_env.iter().enumerate() {
-                pass_thru_env_builder.set(i as u32, env);
+                pass_through_env_builder.set(i as u32, env);
             }
         }
 
@@ -307,7 +315,9 @@ impl<'a> From<GlobalHashable<'a>> for Builder<HeapAllocator> {
             }
         }
 
-        builder.set_root_external_deps_hash(&hashable.root_external_dependencies_hash);
+        if let Some(root_external_dependencies_hash) = hashable.root_external_dependencies_hash {
+            builder.set_root_external_deps_hash(&root_external_dependencies_hash);
+        }
 
         {
             let mut entries = builder.reborrow().init_env(hashable.env.len() as u32);
@@ -384,9 +394,9 @@ mod test {
         let task_hashable = TaskHashable {
             global_hash: "global_hash",
             task_dependency_hashes: vec!["task_dependency_hash".to_string()],
-            package_dir: turbopath::RelativeUnixPathBuf::new("package_dir").unwrap(),
+            package_dir: Some(turbopath::RelativeUnixPathBuf::new("package_dir").unwrap()),
             hash_of_files: "hash_of_files",
-            external_deps_hash: "external_deps_hash".to_string(),
+            external_deps_hash: Some("external_deps_hash".to_string()),
             task: "task",
             outputs: TaskOutputs {
                 inclusions: vec!["inclusions".to_string()],
@@ -413,7 +423,7 @@ mod test {
             )]
             .into_iter()
             .collect(),
-            root_external_dependencies_hash: "0000000000000000".to_string(),
+            root_external_dependencies_hash: Some("0000000000000000".to_string()),
             env: &["env".to_string()],
             resolved_env_vars: vec![],
             pass_through_env: &["pass_through_env".to_string()],