Move Watchman integration into the library

Summary: Want things using open source to be able to use watchman as well.

Reviewed By: SamChou19815

Differential Revision: D73776774

fbshipit-source-id: 10225d73087068423d813f899ab2926dff4bd5ba

Author: ndmitchell

pyrefly/Cargo.toml

@@ -75,6 +75,7 @@ tar = "0.4.44"
 zstd = { version = "0.13", features = ["experimental", "zstdmt"] }
 
 [target.'cfg(not(target_arch = "wasm32"))'.dependencies]
+watchman_client = "0.9.0"
 which = "4.2.4"
 zstd = { version = "0.13", features = ["experimental", "zstdmt"] }
 

pyrefly/lib/lib.rs

@@ -77,10 +77,14 @@ pub mod library {
                 pub use crate::util::args::clap_env;
                 pub use crate::util::args::get_args_expanded;
                 pub use crate::util::globs;
+                #[cfg(not(target_arch = "wasm32"))]
                 pub use crate::util::notify_watcher::NotifyWatcher;
                 pub use crate::util::thread_pool::init_thread_pool;
                 pub use crate::util::trace::init_tracing;
+                #[cfg(not(target_arch = "wasm32"))]
                 pub use crate::util::watcher::Watcher;
+                #[cfg(not(target_arch = "wasm32"))]
+                pub use crate::util::watchman::Watchman;
             }
         }
     }

pyrefly/lib/util/mod.rs

@@ -20,6 +20,7 @@ pub mod lock;
 pub mod locked_map;
 pub mod memory;
 pub mod no_hash;
+#[cfg(not(target_arch = "wasm32"))]
 pub mod notify_watcher;
 pub mod prelude;
 pub mod recurser;
@@ -32,4 +33,6 @@ pub mod upgrade_lock;
 pub mod upward_search;
 pub mod visit;
 pub mod watcher;
+#[cfg(not(target_arch = "wasm32"))]
+pub mod watchman;
 pub mod with_hash;

pyrefly/lib/util/watchman.rs

@@ -0,0 +1,163 @@
+/*
+ * Copyright (c) Meta Platforms, Inc. and affiliates.
+ *
+ * This source code is licensed under the MIT license found in the
+ * LICENSE file in the root directory of this source tree.
+ */
+
+use std::path::Path;
+use std::path::PathBuf;
+
+use anyhow::Context as _;
+use async_trait::async_trait;
+use notify::Event;
+use tracing::debug;
+use tracing::info;
+use watchman_client::CanonicalPath;
+use watchman_client::Connector;
+use watchman_client::Subscription;
+use watchman_client::SubscriptionData;
+use watchman_client::expr::Expr;
+use watchman_client::pdu::Clock;
+use watchman_client::pdu::FileType;
+use watchman_client::pdu::SubscribeRequest;
+use watchman_client::pdu::SyncTimeout;
+
+use crate::util::watcher::Watcher;
+
+// The "new" field is marked as deprecated, but buck2 uses it and
+// I'm unaware of issues due to its use there.
+//
+// Putting this in it own mod was the best way to scope the allow(deprecated).
+#[allow(deprecated)]
+mod watchman_query {
+    use std::path::Path;
+
+    use notify::Event;
+    use notify::EventKind;
+    use notify::event::CreateKind;
+    use notify::event::EventAttributes;
+    use notify::event::ModifyKind;
+    use notify::event::RemoveKind;
+    use serde::Deserialize;
+    use watchman_client::prelude::*;
+
+    query_result_type! {
+        pub struct SubscriptionFields {
+            name: NameField,
+            file_type: FileTypeField,
+            exists: ExistsField,
+            new: NewField,
+        }
+    }
+
+    impl SubscriptionFields {
+        pub fn into_event(self, root: &Path) -> Option<Event> {
+            match *self.file_type {
+                FileType::BlockSpecial
+                | FileType::CharSpecial
+                | FileType::Fifo
+                | FileType::Socket
+                | FileType::SolarisDoor
+                | FileType::Unknown => {
+                    return None;
+                }
+                FileType::Directory | FileType::Regular | FileType::Symlink => {}
+            };
+
+            let kind = match (*self.exists, *self.new) {
+                (true, true) => EventKind::Create(CreateKind::Any),
+                (false, _) => EventKind::Remove(RemoveKind::Any),
+                (true, false) => EventKind::Modify(ModifyKind::Any),
+            };
+
+            Some(Event {
+                kind,
+                paths: vec![root.join(self.name.as_path())],
+                attrs: EventAttributes::new(),
+            })
+        }
+    }
+}
+
+pub struct Watchman {
+    root: PathBuf,
+    subscription: Subscription<watchman_query::SubscriptionFields>,
+}
+
+#[async_trait]
+impl Watcher for Watchman {
+    async fn wait(&mut self) -> anyhow::Result<Vec<Event>> {
+        loop {
+            match self
+                .subscription
+                .next()
+                .await
+                .context("Failed to get watchman response")?
+            {
+                SubscriptionData::Canceled => {
+                    return Err(anyhow::anyhow!("Watchman subscription was canceled"));
+                }
+                SubscriptionData::StateEnter { .. } | SubscriptionData::StateLeave { .. } => {
+                    continue;
+                }
+                SubscriptionData::FilesChanged(changes) => {
+                    if let Some(files) = changes.files {
+                        info!("Received watchman changes on {} files", files.len());
+                        return Ok(files
+                            .into_iter()
+                            .filter_map(|file| file.into_event(&self.root))
+                            .collect());
+                    } else {
+                        continue;
+                    }
+                }
+            }
+        }
+    }
+}
+
+impl Watchman {
+    pub async fn new(path: &Path) -> anyhow::Result<Self> {
+        let root = CanonicalPath::canonicalize(path)?;
+        let watchman_client = Connector::new()
+            .connect()
+            .await
+            .context("Failed to connect to watchman client")?;
+        let root = watchman_client
+            .resolve_root(root)
+            .await
+            .context("Failed to resolve root directory for watchman client")?;
+        let clock = watchman_client
+            .clock(&root, SyncTimeout::Default)
+            .await
+            .context("Failed to obtain initial clock from watchman client")?;
+        let (subscription, initial_response) = watchman_client
+            .subscribe::<watchman_query::SubscriptionFields>(
+                &root,
+                SubscribeRequest {
+                    // If `empty_on_fresh_instance` is false (default), and if `since` is None,
+                    // watchman is going to crawl & return the entire file tree on the first subscription
+                    // response, which can be very slow. Explicitly setting the `since` clock can
+                    // significantly speed up the setup.
+                    since: Some(Clock::Spec(clock)),
+                    relative_root: Some(PathBuf::from(".")),
+                    expression: Some(Expr::All(vec![
+                        Expr::FileType(FileType::Regular),
+                        Expr::Suffix(vec![PathBuf::from("py"), PathBuf::from("pyi")]),
+                    ])),
+                    ..SubscribeRequest::default()
+                },
+            )
+            .await
+            .context("Failed to subscribe to changes")?;
+        debug!(
+            "Successfully established watchman subscription. Initial response: {:#?}",
+            initial_response
+        );
+        Ok(Watchman {
+            root: root.path(),
+            subscription,
+        })
+    }
+}