Repository: vercel/next.js
Pull request: fix(turbo-tasks-fs): Handle filesystem watcher rescan events

The notify crate requires handling rescan events with the event queue overflows.

Context on this issue in the notify crate:
- https://github.com/notify-rs/notify/issues/412
- https://github.com/notify-rs/notify/issues/434
- https://github.com/notify-rs/notify/pull/435

In our case, that means we should invalidate every file we've read in `turbo-tasks-fs`.

We should also attempt to re-add filesystem watchers for every path, since some of those paths could've been removed and re-added.

Tested this on Linux by lowering the inotify queue to something tiny to trigger frequent rescan events:

```
echo 3 | sudo tee /proc/sys/fs/inotify/max_queued_events
rm -rf /tmp/fuzz && RUST_BACKTRACE=1 cargo run -p turbo-tasks-fuzz -- fs-watcher --fs-root /tmp/fuzz --notify-timeout-ms 100 --directory-modifications 0
echo 16384 | sudo tee /proc/sys/fs/inotify/max_queued_events  # restore back to default
```

As a sanity check, I also tried running the fuzzer on macos, though it doesn't seem like there's as easy of a way to trigger rescan events on that platform (?).

Closes PACK-4394

Changed files:
filename: crates/next-api/src/project.rs
patch: @@ -33,7 +33,7 @@ use turbo_tasks::{
     ReadRef, ResolvedVc, State, TaskInput, TransientInstance, TryFlatJoinIterExt, Value, Vc,
 };
 use turbo_tasks_env::{EnvMap, ProcessEnv};
-use turbo_tasks_fs::{DiskFileSystem, FileSystem, FileSystemPath, VirtualFileSystem};
+use turbo_tasks_fs::{invalidation, DiskFileSystem, FileSystem, FileSystemPath, VirtualFileSystem};
 use turbopack::{
     evaluate_context::node_build_environment, global_module_ids::get_global_module_id_strategy,
     transition::TransitionOptions, ModuleAssetContext,
@@ -308,12 +308,16 @@ impl ProjectContainer {
                 .start_watching_with_invalidation_reason(watch.poll_interval)
                 .await?;
         } else {
-            project_fs.invalidate_with_reason();
+            project_fs.invalidate_with_reason(|path| invalidation::Initialize {
+                path: RcStr::from(path),
+            });
         }
         let output_fs = output_fs_operation(project)
             .read_strongly_consistent()
             .await?;
-        output_fs.invalidate_with_reason();
+        output_fs.invalidate_with_reason(|path| invalidation::Initialize {
+            path: RcStr::from(path),
+        });
         Ok(())
     }
 
@@ -402,11 +406,15 @@ impl ProjectContainer {
                     .start_watching_with_invalidation_reason(watch.poll_interval)
                     .await?;
             } else {
-                project_fs.invalidate_with_reason();
+                project_fs.invalidate_with_reason(|path| invalidation::Initialize {
+                    path: RcStr::from(path),
+                });
             }
         }
         if !ReadRef::ptr_eq(&prev_output_fs, &output_fs) {
-            prev_output_fs.invalidate_with_reason();
+            prev_output_fs.invalidate_with_reason(|path| invalidation::Initialize {
+                path: RcStr::from(path),
+            });
         }
 
         Ok(())

filename: turbopack/crates/turbo-tasks-fs/src/invalidation.rs
patch: @@ -8,7 +8,7 @@ use turbo_tasks::{util::StaticOrArc, FxIndexSet, InvalidationReason, Invalidatio
 
 /// Invalidation was caused by a file change detected by the file watcher
 #[derive(PartialEq, Eq, Hash)]
-pub struct WatchChange {
+pub(crate) struct WatchChange {
     pub path: String,
 }
 
@@ -52,7 +52,7 @@ impl InvalidationReasonKind for WatchChangeKind {
 /// Invalidation was caused by a directory starting to watch from which was read
 /// before.
 #[derive(PartialEq, Eq, Hash, Clone)]
-pub struct WatchStart {
+pub(crate) struct WatchStart {
     pub name: RcStr,
     pub path: RcStr,
 }
@@ -84,92 +84,93 @@ impl InvalidationReasonKind for WatchStartKind {
         let example = reasons[0].as_any().downcast_ref::<WatchStart>().unwrap();
         write!(
             f,
-            "{} items started watching (e. g. {} in {})",
+            "{} items started watching (e.g. {} in {})",
             reasons.len(),
             example.path,
             example.name
         )
     }
 }
 
-/// Invalidation was caused by a write operation on the filesystem
-#[derive(PartialEq, Eq, Hash)]
-pub struct Write {
-    pub path: String,
+/// Invalidation was caused by initialization of a project or filesystem.
+#[derive(PartialEq, Eq, Hash, Clone)]
+pub struct Initialize {
+    pub path: RcStr,
 }
 
-impl InvalidationReason for Write {
+impl InvalidationReason for Initialize {
     fn kind(&self) -> Option<StaticOrArc<dyn InvalidationReasonKind>> {
-        Some(StaticOrArc::Static(&WRITE_KIND))
+        Some(StaticOrArc::Static(&INITIALIZE_KIND))
     }
 }
 
-impl Display for Write {
+impl Display for Initialize {
     fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
-        write!(f, "{} written", self.path)
+        write!(
+            f,
+            "initialized project or filesystem with path {}",
+            self.path
+        )
     }
 }
 
-/// Invalidation kind for [Write]
+/// [Invalidation kind][InvalidationReasonKind] for [`Initialize`].
 #[derive(PartialEq, Eq, Hash)]
-struct WriteKind;
+struct InitializeKind;
 
-static WRITE_KIND: WriteKind = WriteKind;
+static INITIALIZE_KIND: InitializeKind = InitializeKind;
 
-impl InvalidationReasonKind for WriteKind {
+impl InvalidationReasonKind for InitializeKind {
     fn fmt(
         &self,
         reasons: &FxIndexSet<StaticOrArc<dyn InvalidationReason>>,
         f: &mut Formatter<'_>,
     ) -> std::fmt::Result {
+        let example = reasons[0].as_any().downcast_ref::<Initialize>().unwrap();
         write!(
             f,
-            "{} files written ({}, ...)",
+            "{} items invalidated as part of project or filesystem initialization ({}, ...)",
             reasons.len(),
-            reasons[0].as_any().downcast_ref::<Write>().unwrap().path
+            example.path,
         )
     }
 }
 
-/// Invalidation was caused by a invalidate operation on the filesystem
-#[derive(Clone, PartialEq, Eq, Hash)]
-pub struct InvalidateFilesystem {
-    pub path: RcStr,
+/// Invalidation was caused by a write operation on the filesystem
+#[derive(PartialEq, Eq, Hash)]
+pub(crate) struct Write {
+    pub path: String,
 }
 
-impl InvalidationReason for InvalidateFilesystem {
+impl InvalidationReason for Write {
     fn kind(&self) -> Option<StaticOrArc<dyn InvalidationReasonKind>> {
-        Some(StaticOrArc::Static(&INVALIDATE_FILESYSTEM_KIND))
+        Some(StaticOrArc::Static(&WRITE_KIND))
     }
 }
 
-impl Display for InvalidateFilesystem {
+impl Display for Write {
     fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
-        write!(f, "{} in filesystem invalidated", self.path)
+        write!(f, "{} written", self.path)
     }
 }
 
-/// Invalidation kind for [InvalidateFilesystem]
+/// Invalidation kind for [Write]
 #[derive(PartialEq, Eq, Hash)]
-struct InvalidateFilesystemKind;
+struct WriteKind;
 
-static INVALIDATE_FILESYSTEM_KIND: InvalidateFilesystemKind = InvalidateFilesystemKind;
+static WRITE_KIND: WriteKind = WriteKind;
 
-impl InvalidationReasonKind for InvalidateFilesystemKind {
+impl InvalidationReasonKind for WriteKind {
     fn fmt(
         &self,
         reasons: &FxIndexSet<StaticOrArc<dyn InvalidationReason>>,
         f: &mut Formatter<'_>,
     ) -> std::fmt::Result {
         write!(
             f,
-            "{} items in filesystem invalidated ({}, ...)",
+            "{} files written ({}, ...)",
             reasons.len(),
-            reasons[0]
-                .as_any()
-                .downcast_ref::<InvalidateFilesystem>()
-                .unwrap()
-                .path
+            reasons[0].as_any().downcast_ref::<Write>().unwrap().path
         )
     }
 }

filename: turbopack/crates/turbo-tasks-fs/src/invalidator_map.rs
patch: @@ -42,14 +42,22 @@ impl InvalidatorMap {
         Ok(guard)
     }
 
-    #[allow(unused_must_use)]
     pub fn insert(
         &self,
         key: String,
         invalidator: Invalidator,
         write_content: Option<WriteContent>,
     ) {
-        self.queue.push((key, invalidator, write_content));
+        self.queue
+            .push((key, invalidator, write_content))
+            .unwrap_or_else(|err| {
+                let (key, ..) = err.into_inner();
+                // PushError<T> is not Debug
+                panic!(
+                    "failed to push {key:?} queue push should never fail, queue is unbounded and \
+                     never closed"
+                )
+            });
     }
 }
 

filename: turbopack/crates/turbo-tasks-fs/src/lib.rs
patch: @@ -11,7 +11,7 @@
 pub mod attach;
 pub mod embed;
 pub mod glob;
-mod invalidation;
+pub mod invalidation;
 mod invalidator_map;
 pub mod json;
 mod mutex_map;
@@ -42,7 +42,6 @@ use bitflags::bitflags;
 use dunce::simplified;
 use glob::Glob;
 use indexmap::IndexSet;
-use invalidation::InvalidateFilesystem;
 use invalidator_map::InvalidatorMap;
 use jsonc_parser::{parse_to_serde_value, ParseOptions};
 use mime::Mime;
@@ -60,7 +59,8 @@ use tracing::Instrument;
 use turbo_rcstr::RcStr;
 use turbo_tasks::{
     debug::ValueDebugFormat, effect, mark_session_dependent, mark_stateful, trace::TraceRawVcs,
-    Completion, Invalidator, NonLocalValue, ReadRef, ResolvedVc, ValueToString, Vc,
+    Completion, InvalidationReason, Invalidator, NonLocalValue, ReadRef, ResolvedVc, ValueToString,
+    Vc,
 };
 use turbo_tasks_hash::{
     hash_xxh3_hash128, hash_xxh3_hash64, DeterministicHash, DeterministicHasher,
@@ -318,7 +318,13 @@ impl DiskFileSystemInner {
         });
     }
 
-    fn invalidate_with_reason(&self) {
+    /// Invalidates every tracked file in the filesystem.
+    ///
+    /// Calls the given
+    fn invalidate_with_reason<R: InvalidationReason + Clone>(
+        &self,
+        reason: impl Fn(String) -> R + Sync,
+    ) {
         let _span = tracing::info_span!("invalidate filesystem", path = &*self.root).entered();
         let span = tracing::Span::current();
         let handle = tokio::runtime::Handle::current();
@@ -329,10 +335,10 @@ impl DiskFileSystemInner {
             .chain(dir_invalidator_map.into_par_iter())
             .flat_map(|(path, invalidators)| {
                 let _span = span.clone().entered();
-                let reason = InvalidateFilesystem { path: path.into() };
+                let reason_for_path = reason(path);
                 invalidators
                     .into_par_iter()
-                    .map(move |i| (reason.clone(), i))
+                    .map(move |i| (reason_for_path.clone(), i))
             });
         iter.for_each(|(reason, (invalidator, _))| {
             let _span = span.clone().entered();
@@ -408,8 +414,11 @@ impl DiskFileSystem {
         self.inner.invalidate();
     }
 
-    pub fn invalidate_with_reason(&self) {
-        self.inner.invalidate_with_reason();
+    pub fn invalidate_with_reason<R: InvalidationReason + Clone>(
+        &self,
+        reason: impl Fn(String) -> R + Sync,
+    ) {
+        self.inner.invalidate_with_reason(reason);
     }
 
     pub async fn start_watching(&self, poll_interval: Option<Duration>) -> Result<()> {

filename: turbopack/crates/turbo-tasks-fs/src/watcher.rs
patch: @@ -1,4 +1,5 @@
 use std::{
+    fmt,
     mem::take,
     path::{Path, PathBuf},
     sync::{
@@ -8,7 +9,7 @@ use std::{
     time::Duration,
 };
 
-use anyhow::Result;
+use anyhow::{Context, Result};
 use notify::{
     event::{MetadataKind, ModifyKind, RenameMode},
     Config, EventKind, PollWatcher, RecommendedWatcher, RecursiveMode, Watcher,
@@ -17,7 +18,11 @@ use rayon::iter::{IntoParallelIterator, ParallelIterator};
 use rustc_hash::{FxHashMap, FxHashSet};
 use serde::{Deserialize, Serialize};
 use tracing::instrument;
-use turbo_tasks::{spawn_thread, Invalidator};
+use turbo_rcstr::RcStr;
+use turbo_tasks::{
+    spawn_thread, util::StaticOrArc, FxIndexSet, InvalidationReason, InvalidationReasonKind,
+    Invalidator,
+};
 
 use crate::{
     format_absolute_fs_path,
@@ -66,10 +71,22 @@ impl DiskWatcher {
         }
     }
 
+    /// Called after a rescan in case a previously watched-but-deleted directory was recreated.
+    #[cfg(not(any(target_os = "macos", target_os = "windows")))]
+    pub(crate) fn restore_all_watching(&self, root_path: &Path) {
+        let mut watcher = self.watcher.lock().unwrap();
+        for dir_path in self.watching.iter() {
+            // TODO: Report diagnostics if this error happens
+            let _ = self.start_watching_dir(&mut watcher, &dir_path, root_path);
+        }
+    }
+
+    /// Called when a new directory is found in a parent directory we're watching.
     #[cfg(not(any(target_os = "macos", target_os = "windows")))]
     pub(crate) fn restore_if_watching(&self, dir_path: &Path, root_path: &Path) -> Result<()> {
         if self.watching.contains(dir_path) {
             let mut watcher = self.watcher.lock().unwrap();
+            // TODO: Also restore any watchers for children of this directory
             self.start_watching_dir(&mut watcher, dir_path, root_path)?;
         }
         Ok(())
@@ -87,33 +104,64 @@ impl DiskWatcher {
         Ok(())
     }
 
+    /// Private helper, assumes that the path has already been added to `self.watching`.
     #[cfg(not(any(target_os = "macos", target_os = "windows")))]
     fn start_watching_dir(
         &self,
         watcher: &mut std::sync::MutexGuard<Option<DiskWatcherInternal>>,
         dir_path: &Path,
         root_path: &Path,
     ) -> Result<()> {
-        use anyhow::Context;
+        // HACK: Rewrite NotFound io errors to PathNotFound
+        // This shouldn't ever happen (notify should transform this for us), but even after
+        // https://github.com/notify-rs/notify/pull/611, it seems like there's still some case where
+        // it can occur with inotify on Linux.
+        fn map_notify_err(mut err: notify::Error) -> notify::Error {
+            if let notify::ErrorKind::Io(io_err) = &err.kind {
+                if io_err.kind() == std::io::ErrorKind::NotFound {
+                    err.kind = notify::ErrorKind::PathNotFound;
+                }
+            }
+            err
+        }
 
         if let Some(watcher) = watcher.as_mut() {
             let mut path = dir_path;
-            while let Err(err) = watcher.watch(path, RecursiveMode::NonRecursive) {
-                if path == root_path {
-                    return Err(err).context(format!(
-                        "Unable to watch {} (tried up to {})",
-                        dir_path.display(),
-                        path.display()
-                    ));
+            let err_with_context = |err| {
+                return Err(err).context(format!(
+                    "Unable to watch {} (tried up to {})",
+                    dir_path.display(),
+                    path.display()
+                ));
+            };
+            while let Err(err) = watcher
+                .watch(path, RecursiveMode::NonRecursive)
+                .map_err(map_notify_err)
+            {
+                match err {
+                    notify::Error {
+                        kind: notify::ErrorKind::PathNotFound,
+                        ..
+                    } => {
+                        // The path was probably deleted before we could process the event. That's
+                        // okay, just make sure we're watching the parent directory, so we can know
+                        // if it gets recreated.
+                        let Some(parent_path) = path.parent() else {
+                            // this should never happen as we break before we reach the root path
+                            return err_with_context(err);
+                        };
+                        if parent_path == root_path {
+                            // assume there's already a root watcher
+                            break;
+                        }
+                        if !self.watching.insert(parent_path.to_owned()) {
+                            // we're already watching the parent path!
+                            break;
+                        }
+                        path = parent_path;
+                    }
+                    _ => return err_with_context(err),
                 }
-                let Some(parent_path) = path.parent() else {
-                    return Err(err).context(format!(
-                        "Unable to watch {} (tried up to {})",
-                        dir_path.display(),
-                        path.display()
-                    ));
-                };
-                path = parent_path;
             }
         }
         Ok(())
@@ -161,8 +209,8 @@ impl DiskWatcher {
             DiskWatcherInternal::Recommended(RecommendedWatcher::new(tx, Config::default())?)
         };
 
-        // Add a path to be watched. All files and directories at that path and
-        // below will be monitored for changes.
+        // Macos and Windows provide efficient recursive directory watchers. On other platforms, we
+        // only track the directories we need: https://github.com/vercel/turborepo/pull/4100
         #[cfg(any(target_os = "macos", target_os = "windows"))]
         {
             watcher.watch(inner.root_path(), RecursiveMode::Recursive)?;
@@ -252,11 +300,52 @@ impl DiskWatcher {
         let mut batched_new_paths = FxHashSet::default();
 
         'outer: loop {
-            let mut event = rx.recv().or(Err(TryRecvError::Disconnected));
+            let mut event_result = rx.recv().or(Err(TryRecvError::Disconnected));
+            // this inner loop batches events using `try_recv`
             loop {
-                match event {
-                    Ok(Ok(notify::Event { kind, paths, .. })) => {
-                        let paths: Vec<PathBuf> = paths
+                match event_result {
+                    Ok(Ok(event)) => {
+                        // TODO: We might benefit from some user-facing diagnostics if it rescans
+                        // occur frequently (i.e. more than X times in Y minutes)
+                        //
+                        // You can test rescans on Linux by reducing the inotify queue to something
+                        // really small:
+                        //
+                        // ```
+                        // echo 3 | sudo tee /proc/sys/fs/inotify/max_queued_events
+                        // ```
+                        if event.need_rescan() {
+                            let _lock = inner.invalidation_lock.blocking_write();
+
+                            #[cfg(not(any(target_os = "macos", target_os = "windows")))]
+                            {
+                                // we can't narrow this down to a smaller set of paths: Rescan
+                                // events (at least when tested on Linux) come with no `paths`, and
+                                // we use only one global `notify::Watcher` instance.
+                                self.restore_all_watching(inner.root_path());
+                                batched_new_paths.clear();
+                            }
+
+                            if report_invalidation_reason {
+                                inner.invalidate_with_reason(|path| InvalidateRescan {
+                                    path: RcStr::from(path),
+                                });
+                            } else {
+                                inner.invalidate();
+                            }
+
+                            // no need to process the rest of the batch as we just
+                            // invalidated everything
+                            batched_invalidate_path.clear();
+                            batched_invalidate_path_dir.clear();
+                            batched_invalidate_path_and_children.clear();
+                            batched_invalidate_path_and_children_dir.clear();
+
+                            break;
+                        }
+
+                        let paths: Vec<PathBuf> = event
+                            .paths
                             .iter()
                             .filter(|p| {
                                 !self
@@ -268,13 +357,15 @@ impl DiskWatcher {
                             .collect();
 
                         if paths.is_empty() {
-                            return;
+                            // this event isn't useful, but keep trying to process the batch
+                            event_result = rx.try_recv();
+                            continue;
                         }
 
                         // [NOTE] there is attrs in the `Event` struct, which contains few
                         // more metadata like process_id who triggered the event,
                         // or the source we may able to utilize later.
-                        match kind {
+                        match event.kind {
                             // [NOTE] Observing `ModifyKind::Metadata(MetadataKind::Any)` is
                             // not a mistake, fix for PACK-2437.
                             // In here explicitly subscribes to the `ModifyKind::Data` which
@@ -291,7 +382,7 @@ impl DiskWatcher {
                             EventKind::Modify(
                                 ModifyKind::Data(_) | ModifyKind::Metadata(MetadataKind::Any),
                             ) => {
-                                batched_invalidate_path.extend(paths.clone());
+                                batched_invalidate_path.extend(paths);
                             }
                             EventKind::Create(_) => {
                                 batched_invalidate_path_and_children.extend(paths.clone());
@@ -390,20 +481,23 @@ impl DiskWatcher {
                         let delay = Duration::from_millis(1);
                         match rx.recv_timeout(delay) {
                             Ok(result) => {
-                                event = Ok(result);
+                                event_result = Ok(result);
                                 continue;
                             }
                             Err(_) => break,
                         }
                     }
                 }
-                event = rx.try_recv();
+                event_result = rx.try_recv();
             }
 
-            // We need to start watching first before invalidating the changed paths
+            // We need to start watching first before invalidating the changed paths...
+            // This is only needed on platforms we don't do recursive watching on:
+            // https://github.com/vercel/turborepo/pull/4100
             #[cfg(not(any(target_os = "macos", target_os = "windows")))]
             {
                 for path in batched_new_paths.drain() {
+                    // TODO: Report diagnostics if this error happens
                     let _ = self.restore_if_watching(&path, inner.root_path());
                 }
             }
@@ -490,3 +584,47 @@ fn invalidate_path_and_children_execute(
         }
     }
 }
+
+/// Invalidation was caused by a watcher rescan event. This will likely invalidate *every* watched
+/// file.
+#[derive(Clone, PartialEq, Eq, Hash)]
+pub struct InvalidateRescan {
+    path: RcStr,
+}
+
+impl InvalidationReason for InvalidateRescan {
+    fn kind(&self) -> Option<StaticOrArc<dyn InvalidationReasonKind>> {
+        Some(StaticOrArc::Static(&INVALIDATE_RESCAN_KIND))
+    }
+}
+
+impl fmt::Display for InvalidateRescan {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        write!(f, "{} in filesystem invalidated", self.path)
+    }
+}
+
+/// [Invalidation kind][InvalidationReasonKind] for [`InvalidateRescan`].
+#[derive(PartialEq, Eq, Hash)]
+struct InvalidateRescanKind;
+
+static INVALIDATE_RESCAN_KIND: InvalidateRescanKind = InvalidateRescanKind;
+
+impl InvalidationReasonKind for InvalidateRescanKind {
+    fn fmt(
+        &self,
+        reasons: &FxIndexSet<StaticOrArc<dyn InvalidationReason>>,
+        f: &mut fmt::Formatter<'_>,
+    ) -> fmt::Result {
+        write!(
+            f,
+            "{} items in filesystem invalidated due to notify::Watcher rescan event ({}, ...)",
+            reasons.len(),
+            reasons[0]
+                .as_any()
+                .downcast_ref::<InvalidateRescan>()
+                .unwrap()
+                .path
+        )
+    }
+}