fix(daemon): kill daemon when root is removed

crates/turborepo-globwatch/src/lib.rs

@@ -235,6 +235,8 @@ pub enum ConfigError {
     ServerStopped,
     /// Watch error
     WatchError(Vec<notify::Error>),
+    /// The server has already been consumed.
+    WatchingAlready,
 }
 
 impl<T: Watcher> WatchConfig<T> {
@@ -287,6 +289,24 @@ impl<T: Watcher> WatchConfig<T> {
             .map_err(ConfigError::WatchError)
     }
 
+    /// Register a single path to be included by the watcher.
+    pub async fn include_path(&self, path: &Path) -> Result<(), ConfigError> {
+        trace!("watching {:?}", path);
+        // Windows doesn't create an event when a watched directory itself is deleted
+        // we watch the parent directory instead.
+        // More information at https://github.com/notify-rs/notify/issues/403
+        #[cfg(windows)]
+        let watched_path = path.parent().expect("turbo is unusable at filesytem root");
+        #[cfg(not(windows))]
+        let watched_path = path;
+
+        self.watcher
+            .lock()
+            .expect("watcher lock poisoned")
+            .watch(watched_path, notify::RecursiveMode::NonRecursive)
+            .map_err(|e| ConfigError::WatchError(vec![e]))
+    }
+
     /// Register a glob to be excluded by the watcher.
     #[tracing::instrument(skip(self))]
     pub async fn exclude(&self, relative_to: &Path, glob: &str) {

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

@@ -195,7 +195,15 @@ impl<T: Watcher + Send + 'static> DaemonServer<T> {
                     None => CloseReason::ServerClosed,
                 }
             },
-            _ = watcher_fut => CloseReason::WatcherClosed,
+            watch_res = watcher_fut => {
+                match watch_res {
+                    Ok(()) => CloseReason::WatcherClosed,
+                    Err(e) => {
+                        error!("Globwatch config error: {:?}", e);
+                        CloseReason::WatcherClosed
+                    },
+                }
+            },
         }
 
         // here the stop token is dropped, and the pid lock is dropped

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

@@ -8,7 +8,7 @@ use std::{
 use futures::{stream::iter, StreamExt};
 use globwatch::{ConfigError, GlobWatcher, StopToken, WatchConfig, Watcher};
 use itertools::Itertools;
-use notify::RecommendedWatcher;
+use notify::{EventKind, RecommendedWatcher};
 use tokio::time::timeout;
 use tracing::{trace, warn};
 use turbopath::AbsoluteSystemPathBuf;
@@ -66,7 +66,7 @@ impl<T: Watcher> HashGlobWatcher<T> {
     /// Watches a given path, using the flush_folder as temporary storage to
     /// make sure that file events are handled in the appropriate order.
     #[tracing::instrument(skip(self, token))]
-    pub async fn watch(&self, token: StopToken) {
+    pub async fn watch(&self, token: StopToken) -> Result<(), ConfigError> {
         let start_globs = {
             let lock = self.hash_globs.lock().expect("only fails if poisoned");
             lock.iter()
@@ -79,17 +79,25 @@ impl<T: Watcher> HashGlobWatcher<T> {
             Some(watcher) => watcher.into_stream(token),
             None => {
                 warn!("watcher already consumed");
-                return;
+                return Err(ConfigError::WatchingAlready);
             }
         };
 
+        // watch the root of the repo to shut down if the folder is deleted
+        self.config.include_path(&self.relative_to).await?;
+
         // watch all the globs currently in the map
         for glob in start_globs {
             self.config.include(&self.relative_to, &glob).await.ok();
         }
 
         while let Some(Ok(event)) = stream.next().await {
-            trace!("processing event: {:?}", event);
+            if event.paths.contains(&self.relative_to) && matches!(event.kind, EventKind::Remove(_))
+            {
+                // if the root of the repo is deleted, we shut down
+                trace!("repo root was removed, shutting down");
+                break;
+            }
 
             let repo_relative_paths = event
                 .paths
@@ -115,6 +123,8 @@ impl<T: Watcher> HashGlobWatcher<T> {
                 self.config.exclude(&self.relative_to, &glob).await;
             }
         }
+
+        Ok(())
     }
 
     /// registers a hash with a set of globs to watch for changes
@@ -657,4 +667,36 @@ mod test {
             watcher.glob_statuses.lock().unwrap()
         );
     }
+
+    #[tokio::test]
+    #[tracing_test::traced_test]
+    async fn delete_root_kill_daemon() {
+        let dir = setup();
+        let flush = tempdir::TempDir::new("globwatch-flush").unwrap();
+        let watcher = Arc::new(
+            super::HashGlobWatcher::new(
+                AbsoluteSystemPathBuf::new(dir.path()).unwrap(),
+                flush.path().to_path_buf(),
+            )
+            .unwrap(),
+        );
+
+        let stop = StopSource::new();
+
+        let task_watcher = watcher.clone();
+        let token = stop.token();
+
+        // dropped when the test ends
+        let task = tokio::task::spawn(async move { task_watcher.watch(token).await });
+
+        watcher.config.flush().await.unwrap();
+        std::fs::remove_dir_all(dir.path()).unwrap();
+
+        // it should shut down
+        let finish = task.await;
+        assert!(
+            finish.is_ok(),
+            "expected task to finish when root is deleted"
+        );
+    }
 }