feat(turborepo): Process package change events asynchronously (#8036)

### Description

We were previously processing the package change events synchronously,
which ended up creating too much lagging and therefore crashes. Now we
spin up a thread that reads in the package change events, and
accumulates the changed packages. A separate thread handles execution by
taking those changed packages and spinning up a new run.

Also I changed our logic to just send rediscover events on lagging with
the channel in the daemon server. No reason to crash the server in that
case.

The first commit is more of a refactor to put the watch state into
`WatchClient`, to clean up the state ahead of splitting into two
threads. You can review commit by commit if you wish.

### Testing Instructions

Tested on `next.js`, lagging is definitely improved.

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 {