Shut down watchers before exiting

Author: pasunboneleve

src/core.rs

@@ -109,6 +109,7 @@ pub enum RuntimeEffect {
     LogInfo {
         message: String,
     },
+    StopWatching,
     StopAllProcesses,
     Exit,
 }
@@ -377,6 +378,7 @@ impl RuntimeMachine {
                 self.pending_effects.push_back(RuntimeEffect::LogInfo {
                     message: "received ctrl-c, shutting down".into(),
                 });
+                self.pending_effects.push_back(RuntimeEffect::StopWatching);
                 self.pending_effects
                     .push_back(RuntimeEffect::StopAllProcesses);
                 self.pending_effects.push_back(RuntimeEffect::Exit);
@@ -849,6 +851,7 @@ mod tests {
                 message: "received ctrl-c, shutting down".into()
             })
         );
+        assert_eq!(runtime.next_effect(), Some(RuntimeEffect::StopWatching));
         assert_eq!(runtime.next_effect(), Some(RuntimeEffect::StopAllProcesses));
         assert_eq!(runtime.next_effect(), Some(RuntimeEffect::Exit));
         assert_eq!(runtime.next_effect(), None);

src/engine.rs

@@ -1,5 +1,7 @@
 use std::collections::{BTreeMap, BTreeSet};
 use std::path::Path;
+use std::sync::Arc;
+use std::sync::atomic::{AtomicBool, Ordering};
 use std::sync::mpsc;
 use std::time::Duration;
 
@@ -52,6 +54,7 @@ trait RuntimeEffectAdapter {
     async fn maintain_processes(&mut self) -> Result<()>;
     async fn poll_observed_hook(&mut self, hook: &str) -> Result<bool>;
     async fn log_info(&mut self, message: String) -> Result<()>;
+    async fn stop_watching(&mut self) -> Result<()>;
     async fn stop_all_processes(&mut self) -> Result<()>;
 }
 
@@ -66,6 +69,7 @@ struct LiveRuntimeAdapter<'a, 'b> {
     processes: &'a mut ProcessManager<'b>,
     state: &'a SessionState,
     watcher: &'a mut RecommendedWatcher,
+    watcher_shutdown: Arc<AtomicBool>,
     external_event_tx: tokio::sync::mpsc::UnboundedSender<ExternalEventMessage>,
     external_event_server: Option<ExternalEventServer>,
     browser_reload_server: Option<BrowserReloadServer>,
@@ -88,14 +92,11 @@ impl Engine {
         let (tx, rx) = mpsc::channel();
         let (external_event_tx, mut external_event_rx) = tokio::sync::mpsc::unbounded_channel();
         let tx_watcher = tx.clone();
+        let watcher_shutdown = Arc::new(AtomicBool::new(false));
+        let watcher_shutdown_callback = watcher_shutdown.clone();
         let mut watcher = RecommendedWatcher::new(
             move |result| {
-                if let Err(error) = tx_watcher.send(result) {
-                    error!(
-                        "failed to forward watcher event into runtime loop: {}",
-                        error
-                    );
-                }
+                forward_watcher_event(&tx_watcher, &watcher_shutdown_callback, result);
             },
             NotifyConfig::default(),
         )?;
@@ -111,6 +112,7 @@ impl Engine {
             processes: &mut processes,
             state: &state,
             watcher: &mut watcher,
+            watcher_shutdown,
             external_event_tx,
             external_event_server: None,
             browser_reload_server: None,
@@ -295,6 +297,12 @@ impl RuntimeEffectAdapter for LiveRuntimeAdapter<'_, '_> {
         Ok(())
     }
 
+    async fn stop_watching(&mut self) -> Result<()> {
+        self.watcher_shutdown.store(true, Ordering::Relaxed);
+        self.watcher.unwatch(&self.config.root)?;
+        Ok(())
+    }
+
     async fn stop_all_processes(&mut self) -> Result<()> {
         self.processes.stop_all(self.state).await
     }
@@ -337,6 +345,7 @@ async fn execute_runtime_effects<A: RuntimeEffectAdapter>(
                 }
             }
             RuntimeEffect::LogInfo { message } => adapter.log_info(message).await?,
+            RuntimeEffect::StopWatching => adapter.stop_watching().await?,
             RuntimeEffect::StopAllProcesses => adapter.stop_all_processes().await?,
             RuntimeEffect::Exit => return Ok(true),
         }
@@ -345,6 +354,21 @@ async fn execute_runtime_effects<A: RuntimeEffectAdapter>(
     Ok(false)
 }
 
+fn forward_watcher_event(
+    tx: &mpsc::Sender<notify::Result<Event>>,
+    shutting_down: &AtomicBool,
+    result: notify::Result<Event>,
+) {
+    if let Err(error) = tx.send(result)
+        && !shutting_down.load(Ordering::Relaxed)
+    {
+        error!(
+            "failed to forward watcher event into runtime loop: {}",
+            error
+        );
+    }
+}
+
 #[cfg(test)]
 async fn run_workflow(
     config: &Config,
@@ -903,6 +927,7 @@ mod tests {
         calls: Vec<String>,
         changed_hooks: BTreeMap<String, bool>,
         workflow_errors: BTreeMap<String, String>,
+        watching: bool,
     }
 
     impl MockRuntimeAdapter {
@@ -911,6 +936,7 @@ mod tests {
                 calls: Vec::new(),
                 changed_hooks: BTreeMap::new(),
                 workflow_errors: BTreeMap::new(),
+                watching: false,
             }
         }
     }
@@ -953,6 +979,7 @@ mod tests {
 
         async fn start_watching(&mut self) -> Result<()> {
             self.calls.push("watch".into());
+            self.watching = true;
             Ok(())
         }
 
@@ -971,6 +998,12 @@ mod tests {
             Ok(())
         }
 
+        async fn stop_watching(&mut self) -> Result<()> {
+            self.calls.push("stop_watch".into());
+            self.watching = false;
+            Ok(())
+        }
+
         async fn stop_all_processes(&mut self) -> Result<()> {
             self.calls.push("stop_all".into());
             Ok(())
@@ -1055,11 +1088,29 @@ mod tests {
                 "poll:current_post_slug",
                 "workflow:publish_post_url:",
                 "log:received ctrl-c, shutting down",
+                "stop_watch",
                 "stop_all",
             ]
         );
     }
 
+    #[test]
+    fn forward_watcher_event_ignores_send_failures_after_shutdown() {
+        let (tx, rx) = mpsc::channel();
+        let shutdown = AtomicBool::new(true);
+        drop(rx);
+
+        forward_watcher_event(
+            &tx,
+            &shutdown,
+            Ok(Event {
+                kind: EventKind::Any,
+                paths: vec![PathBuf::from("content/layout.html")],
+                attrs: Default::default(),
+            }),
+        );
+    }
+
     #[tokio::test]
     async fn runtime_machine_does_not_run_observed_workflow_when_hook_state_is_unchanged() {
         let mut config = Config {