Starting a run with most of the logic stubbed out

Cargo.toml

@@ -171,6 +171,7 @@ once_cell = "1.17.1"
 owo-colors = "3.5.0"
 parking_lot = "0.12.1"
 pathdiff = "0.2.1"
+petgraph = "0.6.3"
 pin-project-lite = "0.2.9"
 port_scanner = "0.1.5"
 postcard = "1.0.4"

crates/turborepo-lib/Cargo.toml

@@ -57,6 +57,7 @@ lazy_static = { workspace = true }
 libc = "0.2.140"
 notify = "5.1"
 pidlock = { path = "../turborepo-pidlock" }
+petgraph = { workspace = true }
 prost = "0.11.6"
 reqwest = { workspace = true, default_features = false, features = ["json"] }
 rustc_version_runtime = "0.2.1"
@@ -83,6 +84,7 @@ const_format = "0.2.30"
 go-parse-duration = "0.1.1"
 is-terminal = "0.4.7"
 node-semver = "2.1.0"
+num_cpus = "1.15.0"
 owo-colors.workspace = true
 regex.workspace = true
 tracing-appender = "0.2.2"

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

@@ -21,6 +21,7 @@ pub(crate) mod generate;
 pub(crate) mod link;
 pub(crate) mod login;
 pub(crate) mod logout;
+pub(crate) mod run;
 pub(crate) mod unlink;
 
 pub struct CommandBase {

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

@@ -0,0 +1,24 @@
+use anyhow::Result;
+use tracing::error;
+use turbopath::AbsoluteSystemPathBuf;
+
+use crate::{
+    commands::CommandBase, manager::Manager, opts::Opts, package_json::PackageJson, run::Run,
+};
+
+#[tokio::main]
+async fn run(base: &mut CommandBase) -> Result<()> {
+    // equivalent of optsFromArgs
+    let opts: Opts = (&base.args).try_into()?;
+
+    // equivalent of configureRun
+    let mut run = Run::new(base, opts);
+
+    match run.run().await {
+        Ok(_) => Ok(()),
+        Err(err) => {
+            error!("run failed: {}", err);
+            Err(err)
+        }
+    }
+}

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

@@ -1,5 +1,7 @@
 use serde::{Deserialize, Serialize};
 
+use crate::{opts::RemoteCacheOpts, run::pipeline::Pipeline};
+
 #[derive(Serialize, Deserialize, Debug, Default)]
 #[serde(rename_all = "camelCase")]
 pub struct SpacesJson {
@@ -8,11 +10,15 @@ pub struct SpacesJson {
     pub other: Option<serde_json::Value>,
 }
 
-#[derive(Serialize, Deserialize, Debug)]
+#[derive(Serialize, Deserialize, Default, Debug)]
 #[serde(rename_all = "camelCase")]
 pub struct TurboJson {
     #[serde(flatten)]
     other: serde_json::Value,
+    // This is 'static just to get stuff working
+    pub(crate) remote_cache_opts: Option<RemoteCacheOpts<'static>>,
+    pub space_id: Option<&'static str>,
+    pub pipeline: Pipeline,
     #[serde(skip_serializing_if = "Option::is_none")]
     pub experimental_spaces: Option<SpacesJson>,
 }

crates/turborepo-lib/src/lib.rs

@@ -7,8 +7,11 @@ mod config;
 mod daemon;
 mod execution_state;
 pub(crate) mod globwatcher;
+mod manager;
+mod opts;
 mod package_json;
 mod package_manager;
+mod run;
 mod shim;
 mod tracing;
 mod ui;

crates/turborepo-lib/src/manager.rs

@@ -0,0 +1,9 @@
+pub struct Manager {
+    // TODO
+}
+
+impl Manager {
+    pub fn new() -> Self {
+        Self {}
+    }
+}

crates/turborepo-lib/src/opts.rs

@@ -0,0 +1,149 @@
+use anyhow::{anyhow, Result};
+
+use crate::{
+    cli::{Command, DryRunMode, EnvMode, LogPrefix, RunArgs},
+    daemon::{DaemonClient, DaemonConnector},
+    Args,
+};
+
+pub struct Opts<'a> {
+    pub cache_opts: CacheOpts<'a>,
+    pub run_opts: RunOpts<'a>,
+    pub runcache_opts: RunCacheOpts,
+    pub scope_opts: ScopeOpts,
+}
+
+#[derive(Default)]
+struct CacheOpts<'a> {
+    override_dir: Option<&'a str>,
+    skip_remote: bool,
+    skip_filesystem: bool,
+    workers: u32,
+    pub(crate) remote_cache_opts: Option<RemoteCacheOpts<'a>>,
+}
+
+impl<'a> From<&'a RunArgs> for CacheOpts<'a> {
+    fn from(run_args: &'a RunArgs) -> Self {
+        CacheOpts {
+            override_dir: run_args.cache_dir.as_deref(),
+            skip_filesystem: run_args.remote_only,
+            workers: run_args.cache_workers,
+            ..CacheOpts::default()
+        }
+    }
+}
+
+#[derive(Default, Clone)]
+pub struct RemoteCacheOpts<'a> {
+    team_id: &'a str,
+    signature: bool,
+}
+
+impl<'a> TryFrom<&'a Args> for Opts<'a> {
+    type Error = anyhow::Error;
+
+    fn try_from(args: &'a Args) -> std::result::Result<Self, Self::Error> {
+        let Some(Command::Run(run_args)) = &args.command else {
+          return Err(anyhow!("Expected run command"))
+        };
+        let run_opts = RunOpts::try_from(run_args.as_ref())?;
+        let cache_opts = CacheOpts::from(run_args.as_ref());
+
+        Ok(Self {
+            run_opts,
+            cache_opts,
+            runcache_opts: RunCacheOpts::default(),
+        })
+    }
+}
+
+#[derive(Debug, Default)]
+pub struct RunCacheOpts {
+    pub(crate) output_watcher: Option<DaemonClient<DaemonConnector>>,
+}
+
+pub struct RunOpts<'a> {
+    tasks: &'a [String],
+    concurrency: u32,
+    parallel: bool,
+    env_mode: EnvMode,
+    profile: Option<&'a str>,
+    continue_on_error: bool,
+    passthrough_args: &'a [String],
+    only: bool,
+    dry_run: bool,
+    pub(crate) dry_run_json: bool,
+    pub graph_dot: bool,
+    graph_file: Option<&'a str>,
+    pub(crate) no_daemon: bool,
+    pub(crate) single_package: bool,
+    log_prefix: Option<LogPrefix>,
+    summarize: Option<Option<bool>>,
+    pub(crate) experimental_space_id: Option<&'a str>,
+}
+
+const DEFAULT_CONCURRENCY: u32 = 10;
+
+impl<'a> TryFrom<&'a RunArgs> for RunOpts<'a> {
+    type Error = anyhow::Error;
+
+    fn try_from(args: &'a RunArgs) -> Result<Self> {
+        let concurrency = args
+            .concurrency
+            .as_deref()
+            .map(parse_concurrency)
+            .transpose()?
+            .unwrap_or(DEFAULT_CONCURRENCY);
+
+        let (graph_dot, graph_file) = match &args.graph {
+            Some(file) if file.is_empty() => (true, None),
+            Some(file) => (false, Some(file.as_str())),
+            None => (false, None),
+        };
+
+        Ok(Self {
+            tasks: args.tasks.as_slice(),
+            log_prefix: args.log_prefix,
+            summarize: args.summarize,
+            experimental_space_id: args.experimental_space_id.as_deref(),
+            env_mode: args.env_mode,
+            concurrency,
+            parallel: args.parallel,
+            profile: args.profile.as_deref(),
+            continue_on_error: args.continue_execution,
+            passthrough_args: args.pass_through_args.as_ref(),
+            only: args.only,
+            no_daemon: args.no_daemon,
+            single_package: args.single_package,
+            graph_dot,
+            graph_file,
+            dry_run_json: matches!(args.dry_run, Some(DryRunMode::Json)),
+            dry_run: args.dry_run.is_some(),
+        })
+    }
+}
+
+fn parse_concurrency(concurrency_raw: &str) -> Result<u32> {
+    if let Some(percent) = concurrency_raw.strip_suffix('%') {
+        let percent = percent.parse::<f64>()?;
+        return if percent > 0.0 && percent.is_finite() {
+            Ok((num_cpus::get() as f64 * percent / 100.0).max(1.0) as u32)
+        } else {
+            Err(anyhow!(
+                "invalid percentage value for --concurrency CLI flag. This should be a percentage \
+                 of CPU cores, between 1% and 100% : {}",
+                percent
+            ))
+        };
+    }
+    match concurrency_raw.parse::<u32>() {
+        Ok(concurrency) if concurrency > 1 => Ok(concurrency),
+        Ok(_) | Err(_) => Err(anyhow!(
+            "invalid value for --concurrency CLI flag. This should be a positive integer greater \
+             than or equal to 1: {}",
+            concurrency_raw
+        )),
+    }
+}
+
+pub struct ScopeOpts {}

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

@@ -0,0 +1,39 @@
+use anyhow::Result;
+use turbopath::AbsoluteSystemPathBuf;
+
+use crate::{package_json::PackageJson, run::graph::WorkspaceCatalog};
+
+pub struct Context {
+    pub workspace_graph: petgraph::Graph<String, String>,
+    pub workspace_infos: WorkspaceCatalog,
+}
+
+impl Context {
+    pub fn build_single_package_graph(_root_package_json: PackageJson) -> Result<Context> {
+        // TODO
+        Ok(Context {
+            workspace_graph: petgraph::Graph::new(),
+            workspace_infos: WorkspaceCatalog::default(),
+        })
+    }
+
+    pub fn build_multi_package_graph(
+        _repo_root: &AbsoluteSystemPathBuf,
+        _root_package_json: &PackageJson,
+    ) -> Result<Context> {
+        // TODO
+        Ok(Context {
+            workspace_graph: petgraph::Graph::new(),
+            workspace_infos: WorkspaceCatalog::default(),
+        })
+    }
+
+    pub fn validate(&self) -> Result<()> {
+        // TODO
+        Ok(())
+    }
+
+    pub fn len(&self) -> usize {
+        self.workspace_graph.node_count()
+    }
+}

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

@@ -0,0 +1,58 @@
+use std::collections::BTreeMap;
+
+use anyhow::Result;
+use turbopath::AbsoluteSystemPathBuf;
+
+use crate::{
+    config::TurboJson,
+    run::pipeline::{Pipeline, TaskDefinition},
+};
+
+pub struct CompleteGraph {
+    // TODO: This should actually be an acyclic graph type
+    // Expresses the dependencies between packages
+    workspace_graph: petgraph::Graph<String, String>,
+    // Config from turbo.json
+    pipeline: Pipeline,
+    // Stores the package.json contents by package name
+    workspace_infos: WorkspaceCatalog,
+    // Hash of all global dependencies
+    global_hash: Option<String>,
+
+    task_definitions: BTreeMap<String, TaskDefinition>,
+    repo_root: AbsoluteSystemPathBuf,
+
+    task_hash_tracker: TaskHashTracker,
+}
+
+impl CompleteGraph {
+    pub fn new(
+        workspace_graph: &petgraph::Graph<String, String>,
+        workspace_infos: &WorkspaceCatalog,
+        repo_root: AbsoluteSystemPathBuf,
+    ) -> Self {
+        Self {
+            workspace_graph,
+            workspace_infos,
+            repo_root,
+            global_hash: None,
+            task_definitions: BTreeMap::new(),
+            task_hash_tracker: TaskHashTracker::default(),
+        }
+    }
+
+    pub fn get_turbo_config_from_workspace(
+        &self,
+        _workspace_name: &str,
+        _is_single_package: bool,
+    ) -> Result<TurboJson> {
+        // TODO
+        Ok(TurboJson::default())
+    }
+}
+
+#[derive(Default)]
+pub struct WorkspaceCatalog {}
+
+#[derive(Default)]
+pub struct TaskHashTracker {}