revert hotreload change

NEXT_CHANGELOG.md

@@ -56,12 +56,6 @@ Now metrics and traces have been separated so that metrics are always comprehens
 
 By [@bryncooke](https://github.com/bryncooke) in https://github.com/apollographql/router/pull/2277
 
-### Replace notify recommended watcher with PollWatcher ([Issue #2245](https://github.com/apollographql/router/issues/2245))
-
-We noticed that we kept receiving issues about hot reload. We tried to fix this a while back by moving from HotWatch to Notify, but there are still issues. The problem appears to be caused by having different mechanisms on different platforms. Switching to the PollWatcher, which offers less sophisticated functionality but is the same on all platforms, should solve these issues at the expense of slightly worse reactiveness.
-
-By [@garypen](https://github.com/garypen) in https://github.com/apollographql/router/pull/2276
-
 ### Keep the error path when redacting subgraph errors ([Issue #1818](https://github.com/apollographql/router/issues/1818))
 
 Error redaction was erasing the error's path, which made it impossible to affect the errors to deferred responses. Now the redacted errors keep the path. Since the response shape for the primary and deferred responses are defined from the API schema, there is no possibility of leaking internal schema information here.

apollo-router/src/files.rs

@@ -1,24 +1,11 @@
 use std::path::Path;
-use std::path::PathBuf;
 use std::time::Duration;
 
 use futures::channel::mpsc;
 use futures::prelude::*;
-use notify::event::DataChange;
-use notify::event::MetadataKind;
-use notify::event::ModifyKind;
-use notify::Config;
-use notify::EventKind;
-use notify::PollWatcher;
 use notify::RecursiveMode;
 use notify::Watcher;
 
-#[cfg(not(test))]
-const DEFAULT_WATCH_DURATION: Duration = Duration::from_secs(3);
-
-#[cfg(test)]
-const DEFAULT_WATCH_DURATION: Duration = Duration::from_millis(100);
-
 /// Creates a stream events whenever the file at the path has changes. The stream never terminates
 /// and must be dropped to finish watching.
 ///
@@ -29,36 +16,17 @@ const DEFAULT_WATCH_DURATION: Duration = Duration::from_millis(100);
 /// returns: impl Stream<Item=()>
 ///
 pub(crate) fn watch(path: &Path) -> impl Stream<Item = ()> {
-    watch_with_duration(path, DEFAULT_WATCH_DURATION)
-}
-
-fn watch_with_duration(path: &Path, duration: Duration) -> impl Stream<Item = ()> {
-    // Due to the vagaries of file watching across multiple platforms, instead of watching the
-    // supplied path (file), we are going to watch the parent (directory) of the path.
-    let mut directory = PathBuf::from(path);
-    let watched_path = directory.clone();
-    directory.pop();
-
     let (mut watch_sender, watch_receiver) = mpsc::channel(1);
-    // We can't use the recommended watcher, because there's just too much variation across
-    // platforms and file systems. We use the Poll Watcher, which is implemented consistently
-    // across all platforms. Less reactive than other mechanisms, but at least it's predictable
-    // across all environments. We compare contents as well, which reduces false positives with
-    // some additional processing burden.
-    let config = Config::default()
-        .with_poll_interval(duration)
-        .with_compare_contents(true);
-    let mut watcher = PollWatcher::new(
-        move |res: Result<notify::Event, notify::Error>| match res {
+    let mut watcher =
+        notify::recommended_watcher(move |res: Result<notify::Event, notify::Error>| match res {
             Ok(event) => {
-                // The two kinds of events of interest to use are writes to the metadata of a
-                // watched file and changes to the data of a watched file
-                if matches!(
-                    event.kind,
-                    EventKind::Modify(ModifyKind::Metadata(MetadataKind::WriteTime))
-                        | EventKind::Modify(ModifyKind::Data(DataChange::Any))
-                ) && event.paths.contains(&watched_path)
-                {
+                // We are only interested in  modify events.
+                // We don't want to lose events and a slow consumer could make us
+                // miss an event notification without a re-send strategy.
+                // If we can't send the event because the channel is full, wait
+                // for a short while and try again. Otherwise, we will panic
+                // because it's a non-recoverable error.
+                if let notify::event::EventKind::Modify(_) = event.kind {
                     loop {
                         match watch_sender.try_send(()) {
                             Ok(_) => break,
@@ -77,14 +45,12 @@ fn watch_with_duration(path: &Path, duration: Duration) -> impl Stream<Item = ()
                     }
                 }
             }
-            Err(e) => tracing::error!("event error: {:?}", e),
-        },
-        config,
-    )
-    .unwrap_or_else(|_| panic!("could not create watch on: {:?}", directory));
+            Err(e) => eprintln!("event error: {:?}", e),
+        })
+        .unwrap_or_else(|_| panic!("could not create watch on: {:?}", path));
     watcher
-        .watch(&directory, RecursiveMode::NonRecursive)
-        .unwrap_or_else(|_| panic!("could not watch: {:?}", directory));
+        .watch(path, RecursiveMode::NonRecursive)
+        .unwrap_or_else(|_| panic!("could not watch: {:?}", path));
     // Tell watchers once they should read the file once,
     // then listen to fs events.
     stream::once(future::ready(()))
@@ -93,7 +59,7 @@ fn watch_with_duration(path: &Path, duration: Duration) -> impl Stream<Item = ()
             // This exists to give the stream ownership of the hotwatcher.
             // Without it hotwatch will get dropped and the stream will terminate.
             // This code never actually gets run.
-            // The ideal would be that hotwatch implements a stream and
+            //The ideal would be that hotwatch implements a stream and
             // therefore we don't need this hackery.
             drop(watcher);
         }))
@@ -116,24 +82,26 @@ pub(crate) mod tests {
     #[test(tokio::test)]
     async fn basic_watch() {
         let (path, mut file) = create_temp_file();
-        let mut watch = watch_with_duration(&path, Duration::from_millis(100));
+        let mut watch = watch(&path);
         // This test can be very racy. Without synchronisation, all
         // we can hope is that if we wait long enough between each
         // write/flush then the future will become ready.
         // Signal telling us we are ready
         assert!(futures::poll!(watch.next()).is_ready());
-        write_and_flush(&mut file, "Some data 1").await;
+        write_and_flush(&mut file, "Some data").await;
         assert!(futures::poll!(watch.next()).is_ready());
-        write_and_flush(&mut file, "Some data 2").await;
+        write_and_flush(&mut file, "Some data").await;
         assert!(futures::poll!(watch.next()).is_ready())
     }
 
+    #[cfg(test)]
     pub(crate) fn create_temp_file() -> (PathBuf, File) {
         let path = temp_dir().join(format!("{}", uuid::Uuid::new_v4()));
         let file = std::fs::File::create(&path).unwrap();
         (path, file)
     }
 
+    #[cfg(test)]
     pub(crate) async fn write_and_flush(file: &mut File, contents: &str) {
         file.seek(SeekFrom::Start(0)).unwrap();
         file.set_len(0).unwrap();

apollo-router/src/plugins/rhai.rs

@@ -8,7 +8,6 @@ use std::sync::atomic::AtomicBool;
 use std::sync::atomic::Ordering;
 use std::sync::Arc;
 use std::sync::Mutex;
-use std::time::Duration;
 
 use arc_swap::ArcSwap;
 use futures::future::ready;
@@ -23,12 +22,8 @@ use http::uri::PathAndQuery;
 use http::HeaderMap;
 use http::StatusCode;
 use http::Uri;
-use notify::event::DataChange;
-use notify::event::MetadataKind;
 use notify::event::ModifyKind;
-use notify::Config;
 use notify::EventKind;
-use notify::PollWatcher;
 use notify::RecursiveMode;
 use notify::Watcher;
 use rhai::module_resolvers::FileModuleResolver;
@@ -450,11 +445,8 @@ impl Plugin for Rhai {
 
         let watcher_handle = std::thread::spawn(move || {
             let watching_path = watched_path.clone();
-            let config = Config::default()
-                .with_poll_interval(Duration::from_secs(3))
-                .with_compare_contents(true);
-            let mut watcher = PollWatcher::new(
-                move |res: Result<notify::Event, notify::Error>| {
+            let mut watcher =
+                notify::recommended_watcher(move |res: Result<notify::Event, notify::Error>| {
                     match res {
                         Ok(event) => {
                             // Let's limit the events we are interested in to:
@@ -463,8 +455,7 @@ impl Plugin for Rhai {
                             //  - with suffix "rhai"
                             if matches!(
                                 event.kind,
-                                EventKind::Modify(ModifyKind::Metadata(MetadataKind::WriteTime))
-                                    | EventKind::Modify(ModifyKind::Data(DataChange::Any))
+                                EventKind::Modify(ModifyKind::Data(_))
                                     | EventKind::Create(_)
                                     | EventKind::Remove(_)
                             ) {
@@ -498,10 +489,8 @@ impl Plugin for Rhai {
                         }
                         Err(e) => tracing::error!("rhai watching event error: {:?}", e),
                     }
-                },
-                config,
-            )
-            .unwrap_or_else(|_| panic!("could not create watch on: {:?}", watched_path));
+                })
+                .unwrap_or_else(|_| panic!("could not create watch on: {:?}", watched_path));
             watcher
                 .watch(&watched_path, RecursiveMode::Recursive)
                 .unwrap_or_else(|_| panic!("could not watch: {:?}", watched_path));

apollo-router/src/router.rs

@@ -737,10 +737,8 @@ mod tests {
             UpdateConfiguration(_)
         ));
 
-        // Need different contents, since we won't get an event if content is the same
-        let contents_datadog = include_str!("testdata/datadog.router.yaml");
         // Modify the file and try again
-        write_and_flush(&mut file, contents_datadog).await;
+        write_and_flush(&mut file, contents).await;
         assert!(matches!(
             stream.next().await.unwrap(),
             UpdateConfiguration(_)
@@ -848,17 +846,15 @@ mod tests {
         // First update is guaranteed
         assert!(matches!(stream.next().await.unwrap(), UpdateSchema(_)));
 
-        // Need different contents, since we won't get an event if content is the same
-        let schema_minimal = include_str!("testdata/minimal_supergraph.graphql");
         // Modify the file and try again
-        write_and_flush(&mut file, schema_minimal).await;
+        write_and_flush(&mut file, schema).await;
         assert!(matches!(stream.next().await.unwrap(), UpdateSchema(_)));
     }
 
     #[test(tokio::test)]
     async fn schema_by_file_missing() {
         let mut stream = SchemaSource::File {
-            path: temp_dir().join("does_not_exist"),
+            path: temp_dir().join("does_not_exit"),
             watch: true,
             delay: None,
         }