Move away from the debounced notify watcher (#9754)

### Problem

We're on a slightly older version of the `notify` crate because we wanted to use the "debounced" API, which has a thread that attempts to batch events and dedupe them where possible. But the implementation of debouncing has issues like notify-rs/notify#205.

### Solution

Move away from the debounced watcher towards the trustier direct delivery of events: post #9636 we do our own debouncing via the Graph not double-cleaning `Nodes`, and via the `--loop` delay.

Additionally, fix two cases where code and tests used to race pants' invalidation and assume (intentionally or otherwise) that pants would not notice created files. 

### Result

Reduced risk of lost `notify` events.

[ci skip-jvm-tests]

contrib/go/src/python/pants/contrib/go/tasks/go_buildgen.py

@@ -330,14 +330,21 @@ def _materialize(self, generation_result):
         remote = self.get_options().remote
         existing_go_buildfiles = set()
 
-        def gather_go_buildfiles(rel_path):
-            address_mapper = self.context.address_mapper
-            for build_file in address_mapper.scan_build_files(base_path=rel_path):
-                existing_go_buildfiles.add(build_file)
+        # We scan for existing BUILD files containing Go targets before we begin to materialize
+        # things, because once we have created files (possibly next to existing files), collisions
+        # between the existing definitions and the new definitions are possible.
+        def gather_go_buildfiles(rel_path, is_relevant_target):
+            for build_file in self.context.address_mapper.scan_build_files(rel_path):
+                spec_path = os.path.dirname(build_file)
+                for address in self.context.address_mapper.addresses_in_spec_path(spec_path):
+                    if is_relevant_target(self.context.build_graph.resolve_address(address)):
+                        existing_go_buildfiles.add(address.rel_path)
 
-        gather_go_buildfiles(generation_result.local_root)
+        gather_go_buildfiles(generation_result.local_root, lambda t: isinstance(t, GoLocalSource))
         if remote and generation_result.remote_root != generation_result.local_root:
-            gather_go_buildfiles(generation_result.remote_root)
+            gather_go_buildfiles(
+                generation_result.remote_root, lambda t: isinstance(t, GoRemoteLibrary)
+            )
 
         targets = set(self.context.build_graph.targets(self.is_go))
         if remote and generation_result.remote_root:
@@ -347,22 +354,20 @@ def gather_go_buildfiles(rel_path):
             remote_root = os.path.join(get_buildroot(), generation_result.remote_root)
             targets.update(self.context.scan(remote_root).targets(self.is_remote_lib))
 
+        # Generate targets, and discard any BUILD files that were overwritten.
         failed_results = []
         for result in self.generate_build_files(targets):
             existing_go_buildfiles.discard(result.build_file_path)
             result.log(self.context.log)
             if result.failed:
                 failed_results.append(result)
 
+        # Finally, unlink any BUILD files that were invalidated but not otherwise overwritten.
         if existing_go_buildfiles:
             deleted = []
             for existing_go_buildfile in existing_go_buildfiles:
-                spec_path = os.path.dirname(existing_go_buildfile)
-                for address in self.context.address_mapper.addresses_in_spec_path(spec_path):
-                    target = self.context.build_graph.resolve_address(address)
-                    if isinstance(target, GoLocalSource):
-                        os.unlink(os.path.join(get_buildroot(), existing_go_buildfile))
-                        deleted.append(existing_go_buildfile)
+                os.unlink(os.path.join(get_buildroot(), existing_go_buildfile))
+                deleted.append(existing_go_buildfile)
             if deleted:
                 self.context.log.info(
                     "Deleted the following obsolete BUILD files:\n\t{}".format(

src/python/pants/engine/internals/mapper_test.py

@@ -204,11 +204,6 @@ def test_no_address_no_family(self) -> None:
         build_file = os.path.join(self.build_root, "a/c", "c.BUILD.json")
         with safe_open(build_file, "w") as fp:
             fp.write('{"type_alias": "struct", "name": "c"}')
-
-        # Exists on disk, but not yet in memory.
-        with self.assertRaises(Exception):
-            self.resolve(spec)
-
         self.scheduler.invalidate_files(["a/c"])
 
         # Success.

src/rust/engine/watch/Cargo.toml

@@ -13,11 +13,7 @@ graph = { path = "../graph" }
 hashing = { path = "../hashing" }
 log = "0.4"
 logging = { path = "../logging" }
-# notify is currently an experimental API, we are pinning to https://docs.rs/notify/5.0.0-pre.1/notify/
-# because the latest prerelease at time of writing has removed the debounced watcher which we would like to use.
-# The author suggests they will add the debounced watcher back into the stable 5.0.0 release. When that happens
-# we can move to it.
-notify = { git = "https://github.com/notify-rs/notify", rev = "fba00891d9105e2f581c69fbe415a58cb7966fdd" }
+notify = "5.0.0-pre.2"
 parking_lot = "0.6"
 task_executor = { path = "../task_executor" }
 

src/rust/engine/watch/src/lib.rs

@@ -81,8 +81,14 @@ impl InvalidationWatcher {
     let canonical_build_root =
       std::fs::canonicalize(build_root.as_path()).map_err(|e| format!("{:?}", e))?;
     let (watch_sender, watch_receiver) = crossbeam_channel::unbounded();
-    let mut watcher: RecommendedWatcher = Watcher::new(watch_sender, Duration::from_millis(50))
-      .map_err(|e| format!("Failed to begin watching the filesystem: {}", e))?;
+    let mut watcher: RecommendedWatcher = Watcher::new_immediate(move |ev| {
+      if watch_sender.send(ev).is_err() {
+        // The watch thread shutting down first is ok, because it can exit when the Invalidatable
+        // is dropped.
+        debug!("Watch thread has shutdown, but Watcher is still running.");
+      }
+    })
+    .map_err(|e| format!("Failed to begin watching the filesystem: {}", e))?;
 
     let (liveness_sender, liveness_receiver) = crossbeam_channel::unbounded();