chore(turbo_json): lift $TURBO_DEFAULT$ handling to turbo.json resolution (#10709)

### Description

This PR moves the handling of `$TURBO_DEFAULT$` to be the responsibility
of `turbo.json` resolution.

Previously it was handled by our SCM module which should only be focused
on SCM operations, not magical strings from Turborepo.

The changes to the daemon only affect the daemon input hashing logic
which is currently disabled.

Reviewing each commit independently is suggested.

### Testing Instructions

Added unit tests and existing tests.

Author: chris-olszewski

crates/turborepo-filewatch/src/hash_watcher.rs

@@ -17,7 +17,7 @@ use tokio::{
 use tracing::{debug, trace};
 use turbopath::{AbsoluteSystemPathBuf, AnchoredSystemPath, AnchoredSystemPathBuf};
 use turborepo_repository::discovery::DiscoveryResponse;
-use turborepo_scm::{package_deps::INPUT_INCLUDE_DEFAULT_FILES, Error as SCMError, GitHashes, SCM};
+use turborepo_scm::{Error as SCMError, GitHashes, SCM};
 
 use crate::{
     debouncer::Debouncer,
@@ -41,15 +41,15 @@ pub enum InputGlobs {
 }
 
 impl InputGlobs {
-    pub fn from_raw(mut raw: Vec<String>) -> Result<Self, GlobError> {
+    pub fn from_raw(raw: Vec<String>, include_default: bool) -> Result<Self, GlobError> {
         if raw.is_empty() {
-            Ok(Self::Default)
-        } else if let Some(default_pos) = raw.iter().position(|g| g == INPUT_INCLUDE_DEFAULT_FILES)
-        {
-            raw.remove(default_pos);
-            Ok(Self::DefaultWithExtras(GlobSet::from_raw_unfiltered(raw)?))
+            return Ok(Self::Default);
+        }
+        let glob_set = GlobSet::from_raw_unfiltered(raw)?;
+        if include_default {
+            Ok(Self::DefaultWithExtras(glob_set))
         } else {
-            Ok(Self::Specific(GlobSet::from_raw_unfiltered(raw)?))
+            Ok(Self::Specific(glob_set))
         }
     }
 
@@ -64,12 +64,16 @@ impl InputGlobs {
     fn as_inputs(&self) -> Vec<String> {
         match self {
             InputGlobs::Default => Vec::new(),
-            InputGlobs::DefaultWithExtras(glob_set) => {
-                let mut inputs = glob_set.as_inputs();
-                inputs.push(INPUT_INCLUDE_DEFAULT_FILES.to_string());
-                inputs
+            InputGlobs::DefaultWithExtras(glob_set) | InputGlobs::Specific(glob_set) => {
+                glob_set.as_inputs()
             }
-            InputGlobs::Specific(glob_set) => glob_set.as_inputs(),
+        }
+    }
+
+    pub fn include_default_files(&self) -> bool {
+        match self {
+            InputGlobs::Default | InputGlobs::DefaultWithExtras(..) => true,
+            InputGlobs::Specific(..) => false,
         }
     }
 }
@@ -542,6 +546,7 @@ impl Subscriber {
                     &repo_root,
                     &spec.package_path,
                     &inputs,
+                    spec.inputs.include_default_files(),
                     telemetry,
                 );
                 trace!("hashing complete for {:?}", spec);

crates/turborepo-lib/src/daemon/client.rs

@@ -13,7 +13,10 @@ use super::{
     proto::{DiscoverPackagesResponse, GetFileHashesResponse},
     Paths,
 };
-use crate::daemon::{proto, proto::PackageChangeEvent};
+use crate::{
+    daemon::proto::{self, PackageChangeEvent},
+    task_graph::TaskInputs,
+};
 
 #[derive(Debug, Clone)]
 pub struct DaemonClient<T> {
@@ -160,13 +163,14 @@ impl<T> DaemonClient<T> {
     pub async fn get_file_hashes(
         &mut self,
         package_path: &AnchoredSystemPath,
-        inputs: &[String],
+        inputs: &TaskInputs,
     ) -> Result<GetFileHashesResponse, DaemonError> {
         let response = self
             .client
             .get_file_hashes(proto::GetFileHashesRequest {
                 package_path: package_path.to_string(),
-                input_globs: inputs.to_vec(),
+                input_globs: inputs.globs.to_vec(),
+                include_default: Some(inputs.default),
             })
             .await?
             .into_inner();

crates/turborepo-lib/src/daemon/proto/turbod.proto

@@ -141,6 +141,7 @@ message GetFileHashesRequest {
   // AnchoredSystemPathBuf
   string package_path = 1;
   repeated string input_globs = 2;
+  optional bool include_default = 3;
 }
 
 message GetFileHashesResponse {

crates/turborepo-lib/src/daemon/server.rs

@@ -386,8 +386,9 @@ impl TurboGrpcServiceInner {
         &self,
         package_path: String,
         inputs: Vec<String>,
+        include_default: bool,
     ) -> Result<HashMap<String, String>, RpcError> {
-        let inputs = InputGlobs::from_raw(inputs)?;
+        let inputs = InputGlobs::from_raw(inputs, include_default)?;
         let package_path = AnchoredSystemPathBuf::try_from(package_path.as_str())
             .map_err(|e| RpcError::InvalidAnchoredPath(package_path, e))?;
         let hash_spec = HashSpec {
@@ -549,7 +550,13 @@ impl proto::turbod_server::Turbod for TurboGrpcServiceInner {
     ) -> Result<tonic::Response<proto::GetFileHashesResponse>, tonic::Status> {
         let inner = request.into_inner();
         let file_hashes = self
-            .get_file_hashes(inner.package_path, inner.input_globs)
+            .get_file_hashes(
+                inner.package_path,
+                inner.input_globs,
+                // If an old client attempts to talk to a new server, assume that we should watch
+                // default files
+                inner.include_default.unwrap_or(true),
+            )
             .await?;
         Ok(tonic::Response::new(proto::GetFileHashesResponse {
             file_hashes,

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

@@ -506,11 +506,11 @@ impl ConfigCache {
 
         // empty inputs to get all files
         let inputs: Vec<String> = vec![];
-        let hash_object = match scm.get_package_file_hashes(repo_root, anchored_root, &inputs, None)
-        {
-            Ok(hash_object) => hash_object,
-            Err(_) => return Err(CacheError::ConfigCacheError),
-        };
+        let hash_object =
+            match scm.get_package_file_hashes(repo_root, anchored_root, &inputs, false, None) {
+                Ok(hash_object) => hash_object,
+                Err(_) => return Err(CacheError::ConfigCacheError),
+            };
 
         // return the hash
         Ok(FileHashes(hash_object).hash())

crates/turborepo-lib/src/run/summary/task.rs

@@ -311,13 +311,13 @@ impl From<TaskDefinition> for TaskSummaryTaskDefinition {
         depends_on.sort();
         outputs.sort();
         env.sort();
-        inputs.sort();
+        inputs.globs.sort();
 
         Self {
             outputs,
             cache,
             depends_on,
-            inputs,
+            inputs: inputs.globs,
             output_logs,
             persistent,
             interruptible,

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

@@ -11,40 +11,8 @@ pub use visitor::{Error as VisitorError, Visitor};
 use crate::{
     cli::{EnvMode, OutputLogsMode},
     run::task_id::{TaskId, TaskName},
-    turbo_json::RawTaskDefinition,
 };
 
-// TaskOutputs represents the patterns for including and excluding files from
-// outputs
-#[derive(Clone, Debug, Default, Serialize, Deserialize, PartialEq, Eq)]
-pub struct TaskOutputs {
-    pub inclusions: Vec<String>,
-    pub exclusions: Vec<String>,
-}
-
-impl TaskOutputs {
-    // We consider an empty outputs to be a log output and nothing else
-    pub fn is_empty(&self) -> bool {
-        self.inclusions.len() == 1
-            && self.inclusions[0].ends_with(".log")
-            && self.exclusions.is_empty()
-    }
-
-    pub fn validated_inclusions(&self) -> Result<Vec<ValidatedGlob>, GlobError> {
-        self.inclusions
-            .iter()
-            .map(|i| ValidatedGlob::from_str(i))
-            .collect()
-    }
-
-    pub fn validated_exclusions(&self) -> Result<Vec<ValidatedGlob>, GlobError> {
-        self.exclusions
-            .iter()
-            .map(|e| ValidatedGlob::from_str(e))
-            .collect()
-    }
-}
-
 // Constructed from a RawTaskDefinition
 #[derive(Debug, PartialEq, Clone, Eq)]
 pub struct TaskDefinition {
@@ -70,10 +38,10 @@ pub struct TaskDefinition {
 
     // Inputs indicate the list of files this Task depends on. If any of those files change
     // we can conclude that any cached outputs or logs for this Task should be invalidated.
-    pub(crate) inputs: Vec<String>,
+    pub inputs: TaskInputs,
 
     // OutputMode determines how we should log the output.
-    pub(crate) output_logs: OutputLogsMode,
+    pub output_logs: OutputLogsMode,
 
     // Persistent indicates whether the Task is expected to exit or not
     // Tasks marked Persistent do not exit (e.g. watch mode or dev servers)
@@ -98,6 +66,22 @@ pub struct TaskDefinition {
     pub with: Option<Vec<Spanned<TaskName<'static>>>>,
 }
 
+// TaskOutputs represents the patterns for including and excluding files from
+// outputs
+#[derive(Clone, Debug, Default, Serialize, Deserialize, PartialEq, Eq)]
+pub struct TaskOutputs {
+    pub inclusions: Vec<String>,
+    pub exclusions: Vec<String>,
+}
+
+// Structure for holding the inputs for a task
+#[derive(Debug, PartialEq, Clone, Eq, Default)]
+pub struct TaskInputs {
+    pub globs: Vec<String>,
+    // Set when $TURBO_DEFAULT$ is in inputs
+    pub default: bool,
+}
+
 impl Default for TaskDefinition {
     fn default() -> Self {
         Self {
@@ -118,16 +102,6 @@ impl Default for TaskDefinition {
     }
 }
 
-impl FromIterator<RawTaskDefinition> for RawTaskDefinition {
-    fn from_iter<T: IntoIterator<Item = RawTaskDefinition>>(iter: T) -> Self {
-        iter.into_iter()
-            .fold(RawTaskDefinition::default(), |mut def, other| {
-                def.merge(other);
-                def
-            })
-    }
-}
-
 const LOG_DIR: &str = ".turbo";
 
 impl TaskDefinition {
@@ -190,6 +164,43 @@ impl TaskDefinition {
     }
 }
 
+impl TaskInputs {
+    pub fn new(globs: Vec<String>) -> Self {
+        Self {
+            globs,
+            default: false,
+        }
+    }
+
+    pub fn with_default(mut self, default: bool) -> Self {
+        self.default = default;
+        self
+    }
+}
+
+impl TaskOutputs {
+    // We consider an empty outputs to be a log output and nothing else
+    pub fn is_empty(&self) -> bool {
+        self.inclusions.len() == 1
+            && self.inclusions[0].ends_with(".log")
+            && self.exclusions.is_empty()
+    }
+
+    pub fn validated_inclusions(&self) -> Result<Vec<ValidatedGlob>, GlobError> {
+        self.inclusions
+            .iter()
+            .map(|i| ValidatedGlob::from_str(i))
+            .collect()
+    }
+
+    pub fn validated_exclusions(&self) -> Result<Vec<ValidatedGlob>, GlobError> {
+        self.exclusions
+            .iter()
+            .map(|e| ValidatedGlob::from_str(e))
+            .collect()
+    }
+}
+
 fn task_log_filename(task_name: &str) -> String {
     format!("turbo-{}.log", task_name.replace(':', "$colon$"))
 }

crates/turborepo-lib/src/task_hash.rs

@@ -190,7 +190,8 @@ impl PackageInputsHashes {
                         let local_hash_result = scm.get_package_file_hashes(
                             repo_root,
                             package_path,
-                            &task_definition.inputs,
+                            &task_definition.inputs.globs,
+                            task_definition.inputs.default,
                             Some(scm_telemetry),
                         );
                         match local_hash_result {
@@ -541,7 +542,7 @@ pub fn get_internal_deps_hash(
 
     let file_hashes = package_dirs
         .into_par_iter()
-        .map(|package_dir| scm.get_package_file_hashes::<&str>(root, package_dir, &[], None))
+        .map(|package_dir| scm.get_package_file_hashes::<&str>(root, package_dir, &[], false, None))
         .reduce(
             || Ok(HashMap::new()),
             |acc, hashes| {