feat(turborepo): Process package change events asynchronously

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

@@ -1286,7 +1286,8 @@ pub async fn run(
             event.track_call();
             let base = CommandBase::new(cli_args, repo_root, version, ui);
 
-            WatchClient::start(base, event).await?;
+            let mut client = WatchClient::new(base, event).await?;
+            client.start().await?;
             // We only exit if we get a signal, so we return a non-zero exit code
             return Ok(1);
         }

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

@@ -18,7 +18,7 @@ use semver::Version;
 use thiserror::Error;
 use tokio::{
     select,
-    sync::{mpsc, oneshot},
+    sync::{broadcast::error::RecvError, mpsc, oneshot},
     task::JoinHandle,
 };
 use tokio_stream::wrappers::ReceiverStream;
@@ -594,7 +594,8 @@ impl proto::turbod_server::Turbod for TurboGrpcServiceInner {
             .package_changes_watcher
             .package_changes()
             .await;
-        let (tx, rx) = mpsc::channel(1);
+
+        let (tx, rx) = mpsc::channel(1024);
 
         tx.send(Ok(proto::PackageChangeEvent {
             event: Some(proto::package_change_event::Event::RediscoverPackages(
@@ -607,6 +608,14 @@ impl proto::turbod_server::Turbod for TurboGrpcServiceInner {
         tokio::spawn(async move {
             loop {
                 let event = match package_changes_rx.recv().await {
+                    Err(RecvError::Lagged(_)) => {
+                        warn!("package changes stream lagged");
+                        proto::PackageChangeEvent {
+                            event: Some(proto::package_change_event::Event::RediscoverPackages(
+                                proto::RediscoverPackages {},
+                            )),
+                        }
+                    }
                     Err(err) => proto::PackageChangeEvent {
                         event: Some(proto::package_change_event::Event::Error(
                             proto::PackageChangeError {

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

@@ -142,11 +142,15 @@ impl Engine<Built> {
     /// Creates an instance of `Engine` that only contains tasks that depend on
     /// tasks from a given package. This is useful for watch mode, where we
     /// need to re-run only a portion of the task graph.
-    pub fn create_engine_for_subgraph(&self, changed_package: &PackageName) -> Engine<Built> {
-        let entrypoint_indices: &[petgraph::graph::NodeIndex] = self
-            .package_tasks
-            .get(changed_package)
-            .map_or(&[], |v| &v[..]);
+    pub fn create_engine_for_subgraph(
+        &self,
+        changed_packages: &HashSet<PackageName>,
+    ) -> Engine<Built> {
+        let entrypoint_indices: Vec<_> = changed_packages
+            .iter()
+            .flat_map(|pkg| self.package_tasks.get(pkg))
+            .flatten()
+            .collect();
 
         // We reverse the graph because we want the *dependents* of entrypoint tasks
         let mut reversed_graph = self.task_graph.clone();
@@ -175,7 +179,7 @@ impl Engine<Built> {
                     .iter()
                     .any(|idx| {
                         node_distances
-                            .get(&(*idx, node_idx))
+                            .get(&(**idx, node_idx))
                             .map_or(false, |dist| *dist != i32::MAX)
                     })
                     .then_some(node.clone())
@@ -764,7 +768,8 @@ mod test {
         engine.task_graph.add_edge(b_build_idx, a_build_idx, ());
 
         let engine = engine.seal();
-        let subgraph = engine.create_engine_for_subgraph(&PackageName::from("a"));
+        let subgraph =
+            engine.create_engine_for_subgraph(&[PackageName::from("a")].into_iter().collect());
 
         // Verify that the subgraph only contains tasks from package `a` and the `build`
         // task from package `b`

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

@@ -62,7 +62,7 @@ pub struct RunBuilder {
     // In watch mode, we can have a changed package that we want to serve as an entrypoint.
     // We will then prune away any tasks that do not depend on tasks inside
     // this package.
-    entrypoint_package: Option<PackageName>,
+    entrypoint_packages: Option<HashSet<PackageName>>,
     should_print_prelude_override: Option<bool>,
 }
 
@@ -114,13 +114,13 @@ impl RunBuilder {
             version,
             experimental_ui,
             analytics_sender: None,
-            entrypoint_package: None,
+            entrypoint_packages: None,
             should_print_prelude_override: None,
         })
     }
 
-    pub fn with_entrypoint_package(mut self, entrypoint_package: PackageName) -> Self {
-        self.entrypoint_package = Some(entrypoint_package);
+    pub fn with_entrypoint_packages(mut self, entrypoint_packages: HashSet<PackageName>) -> Self {
+        self.entrypoint_packages = Some(entrypoint_packages);
         self
     }
 
@@ -451,8 +451,8 @@ impl RunBuilder {
 
         // If we have an initial task, we prune out the engine to only
         // tasks that are reachable from that initial task.
-        if let Some(entrypoint_package) = &self.entrypoint_package {
-            engine = engine.create_engine_for_subgraph(entrypoint_package);
+        if let Some(entrypoint_packages) = &self.entrypoint_packages {
+            engine = engine.create_engine_for_subgraph(entrypoint_packages);
         }
 
         if !self.opts.run_opts.parallel {