refactor: reland "preserve ProcState::file_fetcher between restarts" (#17636)

Just some watcher init step that I thought would be "cloned over" but
needs to be done again on reset.

12 files changed, 145 insertions, 49 deletions

diff --git a/cli/cache/mod.rs b/cli/cache/mod.rs
index a52fe5f48..c8fcaa223 100644
--- a/cli/cache/mod.rs
+++ b/cli/cache/mod.rs
@@ -50,12 +50,10 @@ pub struct FetchCacher {
 impl FetchCacher {
   pub fn new(
     emit_cache: EmitCache,
-    file_fetcher: FileFetcher,
+    file_fetcher: Arc<FileFetcher>,
     root_permissions: PermissionsContainer,
     dynamic_permissions: PermissionsContainer,
   ) -> Self {
-    let file_fetcher = Arc::new(file_fetcher);
-
     Self {
       emit_cache,
       dynamic_permissions,
diff --git a/cli/cache/node.rs b/cli/cache/node.rs
index da17633aa..b19772229 100644
--- a/cli/cache/node.rs
+++ b/cli/cache/node.rs
@@ -24,6 +24,7 @@ struct CjsAnalysisData {
   pub reexports: Vec<String>,
 }
 
+#[derive(Clone)]
 pub struct NodeAnalysisCache {
   db_file_path: Option<PathBuf>,
   inner: Arc<Mutex<Option<Option<NodeAnalysisCacheInner>>>>,
diff --git a/cli/cache/parsed_source.rs b/cli/cache/parsed_source.rs
index 30eecf702..6385e7384 100644
--- a/cli/cache/parsed_source.rs
+++ b/cli/cache/parsed_source.rs
@@ -67,6 +67,14 @@ impl ParsedSourceCache {
     }
   }
 
+  pub fn reset_for_file_watcher(&self) -> Self {
+    Self {
+      db_cache_path: self.db_cache_path.clone(),
+      cli_version: self.cli_version.clone(),
+      sources: Default::default(),
+    }
+  }
+
   pub fn get_parsed_source_from_module(
     &self,
     module: &deno_graph::Module,
diff --git a/cli/graph_util.rs b/cli/graph_util.rs
index 30b426667..3c5545a50 100644
--- a/cli/graph_util.rs
+++ b/cli/graph_util.rs
@@ -71,7 +71,7 @@ pub struct GraphData {
 
 impl GraphData {
   /// Store data from `graph` into `self`.
-  pub fn add_graph(&mut self, graph: &ModuleGraph, reload: bool) {
+  pub fn add_graph(&mut self, graph: &ModuleGraph) {
     for graph_import in &graph.imports {
       for dep in graph_import.dependencies.values() {
         for resolved in [&dep.maybe_code, &dep.maybe_type] {
@@ -90,7 +90,7 @@ impl GraphData {
     let mut has_npm_specifier_in_graph = false;
 
     for (specifier, result) in graph.specifiers() {
-      if !reload && self.modules.contains_key(specifier) {
+      if self.modules.contains_key(specifier) {
         continue;
       }
 
@@ -477,7 +477,7 @@ impl GraphData {
 impl From<&ModuleGraph> for GraphData {
   fn from(graph: &ModuleGraph) -> Self {
     let mut graph_data = GraphData::default();
-    graph_data.add_graph(graph, false);
+    graph_data.add_graph(graph);
     graph_data
   }
 }
@@ -549,7 +549,7 @@ pub async fn create_graph_and_maybe_check(
 
   let check_js = ps.options.check_js();
   let mut graph_data = GraphData::default();
-  graph_data.add_graph(&graph, false);
+  graph_data.add_graph(&graph);
   graph_data
     .check(
       &graph.roots,
diff --git a/cli/module_loader.rs b/cli/module_loader.rs
index 8958a9aa6..c7872988b 100644
--- a/cli/module_loader.rs
+++ b/cli/module_loader.rs
@@ -277,7 +277,6 @@ impl ModuleLoader for CliModuleLoader {
         lib,
         root_permissions,
         dynamic_permissions,
-        false,
       )
       .await
     }
diff --git a/cli/proc_state.rs b/cli/proc_state.rs
index d161a3334..c481a4307 100644
--- a/cli/proc_state.rs
+++ b/cli/proc_state.rs
@@ -75,7 +75,7 @@ pub struct ProcState(Arc<Inner>);
 
 pub struct Inner {
   pub dir: DenoDir,
-  pub file_fetcher: FileFetcher,
+  pub file_fetcher: Arc<FileFetcher>,
   pub http_client: HttpClient,
   pub options: Arc<CliOptions>,
   pub emit_cache: EmitCache,
@@ -128,21 +128,60 @@ impl ProcState {
     let ps =
       Self::build_with_sender(cli_options, Some(files_to_watch_sender.clone()))
         .await?;
+    ps.init_watcher();
+    Ok(ps)
+  }
 
-    // Add the extra files listed in the watch flag
-    if let Some(watch_paths) = ps.options.watch_paths() {
-      files_to_watch_sender.send(watch_paths.clone())?;
-    }
+  /// Reset all runtime state to its default. This should be used on file
+  /// watcher restarts.
+  pub fn reset_for_file_watcher(&mut self) {
+    self.0 = Arc::new(Inner {
+      dir: self.dir.clone(),
+      options: self.options.clone(),
+      emit_cache: self.emit_cache.clone(),
+      emit_options_hash: self.emit_options_hash,
+      emit_options: self.emit_options.clone(),
+      file_fetcher: self.file_fetcher.clone(),
+      http_client: self.http_client.clone(),
+      graph_data: Default::default(),
+      lockfile: self.lockfile.clone(),
+      maybe_import_map: self.maybe_import_map.clone(),
+      maybe_inspector_server: self.maybe_inspector_server.clone(),
+      root_cert_store: self.root_cert_store.clone(),
+      blob_store: Default::default(),
+      broadcast_channel: Default::default(),
+      shared_array_buffer_store: Default::default(),
+      compiled_wasm_module_store: Default::default(),
+      parsed_source_cache: self.parsed_source_cache.reset_for_file_watcher(),
+      maybe_resolver: self.maybe_resolver.clone(),
+      maybe_file_watcher_reporter: self.maybe_file_watcher_reporter.clone(),
+      node_analysis_cache: self.node_analysis_cache.clone(),
+      npm_cache: self.npm_cache.clone(),
+      npm_resolver: self.npm_resolver.clone(),
+      cjs_resolutions: Default::default(),
+      progress_bar: self.progress_bar.clone(),
+      node_std_graph_prepared: AtomicBool::new(false),
+    });
+    self.init_watcher();
+  }
 
-    if let Ok(Some(import_map_path)) = ps
+  // Add invariant files like the import map and explicit watch flag list to
+  // the watcher. Dedup for build_for_file_watcher and reset_for_file_watcher.
+  fn init_watcher(&self) {
+    let files_to_watch_sender = match &self.0.maybe_file_watcher_reporter {
+      Some(reporter) => &reporter.sender,
+      None => return,
+    };
+    if let Some(watch_paths) = self.options.watch_paths() {
+      files_to_watch_sender.send(watch_paths.clone()).unwrap();
+    }
+    if let Ok(Some(import_map_path)) = self
       .options
       .resolve_import_map_specifier()
       .map(|ms| ms.and_then(|ref s| s.to_file_path().ok()))
     {
-      files_to_watch_sender.send(vec![import_map_path])?;
+      files_to_watch_sender.send(vec![import_map_path]).unwrap();
     }
-
-    Ok(ps)
   }
 
   async fn build_with_sender(
@@ -236,7 +275,7 @@ impl ProcState {
         .write_hashable(&emit_options)
         .finish(),
       emit_options,
-      file_fetcher,
+      file_fetcher: Arc::new(file_fetcher),
       http_client,
       graph_data: Default::default(),
       lockfile,
@@ -271,7 +310,6 @@ impl ProcState {
     lib: TsTypeLib,
     root_permissions: PermissionsContainer,
     dynamic_permissions: PermissionsContainer,
-    reload_on_watch: bool,
   ) -> Result<(), AnyError> {
     log::debug!("Preparing module load.");
     let _pb_clear_guard = self.progress_bar.clear_guard();
@@ -280,7 +318,7 @@ impl ProcState {
       r.scheme() == "npm" && NpmPackageReference::from_specifier(r).is_ok()
     });
 
-    if !reload_on_watch && !has_root_npm_specifier {
+    if !has_root_npm_specifier {
       let graph_data = self.graph_data.read();
       if self.options.type_check_mode() == TypeCheckMode::None
         || graph_data.is_type_checked(&roots, &lib)
@@ -314,7 +352,6 @@ impl ProcState {
     struct ProcStateLoader<'a> {
       inner: &'a mut cache::FetchCacher,
       graph_data: Arc<RwLock<GraphData>>,
-      reload: bool,
     }
     impl Loader for ProcStateLoader<'_> {
       fn get_cache_info(
@@ -331,9 +368,7 @@ impl ProcState {
         let graph_data = self.graph_data.read();
         let found_specifier = graph_data.follow_redirect(specifier);
         match graph_data.get(&found_specifier) {
-          Some(_) if !self.reload => {
-            Box::pin(futures::future::ready(Err(anyhow!(""))))
-          }
+          Some(_) => Box::pin(futures::future::ready(Err(anyhow!("")))),
           _ => self.inner.load(specifier, is_dynamic),
         }
       }
@@ -341,7 +376,6 @@ impl ProcState {
     let mut loader = ProcStateLoader {
       inner: &mut cache,
       graph_data: self.graph_data.clone(),
-      reload: reload_on_watch,
     };
 
     let maybe_file_watcher_reporter: Option<&dyn deno_graph::source::Reporter> =
@@ -380,7 +414,7 @@ impl ProcState {
 
     let (npm_package_reqs, has_node_builtin_specifier) = {
       let mut graph_data = self.graph_data.write();
-      graph_data.add_graph(&graph, reload_on_watch);
+      graph_data.add_graph(&graph);
       let check_js = self.options.check_js();
       graph_data
         .check(
@@ -479,7 +513,6 @@ impl ProcState {
         lib,
         PermissionsContainer::allow_all(),
         PermissionsContainer::allow_all(),
-        false,
       )
       .await
   }
@@ -493,7 +526,7 @@ impl ProcState {
     let node_std_graph = self
       .create_graph(vec![node::MODULE_ALL_URL.clone()])
       .await?;
-    self.graph_data.write().add_graph(&node_std_graph, false);
+    self.graph_data.write().add_graph(&node_std_graph);
     self.node_std_graph_prepared.store(true, Ordering::Relaxed);
     Ok(())
   }
diff --git a/cli/tests/integration/watcher_tests.rs b/cli/tests/integration/watcher_tests.rs
index 113067122..5ba16cd5e 100644
--- a/cli/tests/integration/watcher_tests.rs
+++ b/cli/tests/integration/watcher_tests.rs
@@ -635,9 +635,14 @@ fn run_watch_external_watch_files() {
 
   // Change content of the external file
   write(&external_file_to_watch, "Hello world2").unwrap();
+  wait_contains("Restarting", &mut stderr_lines);
+  wait_contains("Process finished", &mut stderr_lines);
 
+  // Again (https://github.com/denoland/deno/issues/17584)
+  write(&external_file_to_watch, "Hello world3").unwrap();
   wait_contains("Restarting", &mut stderr_lines);
   wait_contains("Process finished", &mut stderr_lines);
+
   check_alive_then_kill(child);
 }
 
@@ -1086,6 +1091,44 @@ fn test_watch_unload_handler_error_on_drop() {
   check_alive_then_kill(child);
 }
 
+// Regression test for https://github.com/denoland/deno/issues/15465.
+#[test]
+fn run_watch_reload_once() {
+  let _g = util::http_server();
+  let t = TempDir::new();
+  let file_to_watch = t.path().join("file_to_watch.js");
+  let file_content = r#"
+      import { time } from "http://localhost:4545/dynamic_module.ts";
+      console.log(time);
+    "#;
+  write(&file_to_watch, file_content).unwrap();
+
+  let mut child = util::deno_cmd()
+    .current_dir(util::testdata_path())
+    .arg("run")
+    .arg("--watch")
+    .arg("--reload")
+    .arg(&file_to_watch)
+    .env("NO_COLOR", "1")
+    .stdout(std::process::Stdio::piped())
+    .stderr(std::process::Stdio::piped())
+    .spawn()
+    .unwrap();
+  let (mut stdout_lines, mut stderr_lines) = child_lines(&mut child);
+
+  wait_contains("finished", &mut stderr_lines);
+  let first_output = stdout_lines.next().unwrap();
+
+  write(&file_to_watch, file_content).unwrap();
+  // The remote dynamic module should not have been reloaded again.
+
+  wait_contains("finished", &mut stderr_lines);
+  let second_output = stdout_lines.next().unwrap();
+  assert_eq!(second_output, first_output);
+
+  check_alive_then_kill(child);
+}
+
 #[test]
 fn run_watch_dynamic_imports() {
   let t = TempDir::new();
@@ -1147,11 +1190,11 @@ fn run_watch_dynamic_imports() {
     &mut stdout_lines,
   );
 
-  wait_contains("finished", &mut stderr_lines);
   wait_for(
     |m| m.contains("Watching paths") && m.contains("imported2.js"),
     &mut stderr_lines,
   );
+  wait_contains("finished", &mut stderr_lines);
 
   write(
     &file_to_watch3,
diff --git a/cli/tools/bench.rs b/cli/tools/bench.rs
index 18ac63f38..419ac2da6 100644
--- a/cli/tools/bench.rs
+++ b/cli/tools/bench.rs
@@ -30,6 +30,7 @@ use indexmap::IndexMap;
 use log::Level;
 use serde::Deserialize;
 use serde::Serialize;
+use std::cell::RefCell;
 use std::collections::HashSet;
 use std::path::Path;
 use std::path::PathBuf;
@@ -337,7 +338,6 @@ async fn check_specifiers(
     lib,
     PermissionsContainer::allow_all(),
     PermissionsContainer::new(permissions),
-    true,
   )
   .await?;
 
@@ -529,12 +529,14 @@ pub async fn run_benchmarks_with_watch(
     Permissions::from_options(&ps.options.permissions_options())?;
   let no_check = ps.options.type_check_mode() == TypeCheckMode::None;
 
+  let ps = RefCell::new(ps);
+
   let resolver = |changed: Option<Vec<PathBuf>>| {
     let paths_to_watch = bench_options.files.include.clone();
     let paths_to_watch_clone = paths_to_watch.clone();
     let files_changed = changed.is_some();
     let bench_options = &bench_options;
-    let ps = ps.clone();
+    let ps = ps.borrow().clone();
 
     async move {
       let bench_modules =
@@ -638,7 +640,8 @@ pub async fn run_benchmarks_with_watch(
   let operation = |modules_to_reload: Vec<ModuleSpecifier>| {
     let permissions = &permissions;
     let bench_options = &bench_options;
-    let ps = ps.clone();
+    ps.borrow_mut().reset_for_file_watcher();
+    let ps = ps.borrow().clone();
 
     async move {
       let specifiers =
@@ -663,12 +666,13 @@ pub async fn run_benchmarks_with_watch(
     }
   };
 
+  let clear_screen = !ps.borrow().options.no_clear_screen();
   file_watcher::watch_func(
     resolver,
     operation,
     file_watcher::PrintConfig {
       job_name: "Bench".to_string(),
-      clear_screen: !ps.options.no_clear_screen(),
+      clear_screen,
     },
   )
   .await?;
diff --git a/cli/tools/run.rs b/cli/tools/run.rs
index d5b5eb981..2d001c42b 100644
--- a/cli/tools/run.rs
+++ b/cli/tools/run.rs
@@ -1,7 +1,6 @@
 // Copyright 2018-2023 the Deno authors. All rights reserved. MIT license.
 
 use std::io::Read;
-use std::path::PathBuf;
 use std::sync::Arc;
 
 use deno_ast::MediaType;
@@ -103,16 +102,13 @@ async fn run_with_watch(flags: Flags, script: String) -> Result<i32, AnyError> {
   let flags = Arc::new(flags);
   let main_module = resolve_url_or_path(&script)?;
   let (sender, receiver) = tokio::sync::mpsc::unbounded_channel();
+  let mut ps =
+    ProcState::build_for_file_watcher((*flags).clone(), sender.clone()).await?;
 
-  let operation = |(sender, main_module): (
-    tokio::sync::mpsc::UnboundedSender<Vec<PathBuf>>,
-    ModuleSpecifier,
-  )| {
-    let flags = flags.clone();
+  let operation = |main_module: ModuleSpecifier| {
+    ps.reset_for_file_watcher();
+    let ps = ps.clone();
     Ok(async move {
-      let ps =
-        ProcState::build_for_file_watcher((*flags).clone(), sender.clone())
-          .await?;
       let permissions = PermissionsContainer::new(Permissions::from_options(
         &ps.options.permissions_options(),
       )?);
@@ -126,7 +122,7 @@ async fn run_with_watch(flags: Flags, script: String) -> Result<i32, AnyError> {
   util::file_watcher::watch_func2(
     receiver,
     operation,
-    (sender, main_module),
+    main_module,
     util::file_watcher::PrintConfig {
       job_name: "Process".to_string(),
       clear_screen: !flags.no_clear_screen,
diff --git a/cli/tools/test.rs b/cli/tools/test.rs
index 08000c765..e680d5718 100644
--- a/cli/tools/test.rs
+++ b/cli/tools/test.rs
@@ -44,6 +44,7 @@ use rand::seq::SliceRandom;
 use rand::SeedableRng;
 use regex::Regex;
 use serde::Deserialize;
+use std::cell::RefCell;
 use std::collections::BTreeMap;
 use std::collections::HashSet;
 use std::fmt::Write as _;
@@ -955,7 +956,6 @@ pub async fn check_specifiers(
       lib,
       PermissionsContainer::new(Permissions::allow_all()),
       PermissionsContainer::new(permissions.clone()),
-      false,
     )
     .await?;
   }
@@ -977,7 +977,6 @@ pub async fn check_specifiers(
     lib,
     PermissionsContainer::allow_all(),
     PermissionsContainer::new(permissions),
-    true,
   )
   .await?;
 
@@ -1353,12 +1352,14 @@ pub async fn run_tests_with_watch(
     Permissions::from_options(&ps.options.permissions_options())?;
   let no_check = ps.options.type_check_mode() == TypeCheckMode::None;
 
+  let ps = RefCell::new(ps);
+
   let resolver = |changed: Option<Vec<PathBuf>>| {
     let paths_to_watch = test_options.files.include.clone();
     let paths_to_watch_clone = paths_to_watch.clone();
     let files_changed = changed.is_some();
     let test_options = &test_options;
-    let ps = ps.clone();
+    let ps = ps.borrow().clone();
 
     async move {
       let test_modules = if test_options.doc {
@@ -1466,7 +1467,8 @@ pub async fn run_tests_with_watch(
   let operation = |modules_to_reload: Vec<ModuleSpecifier>| {
     let permissions = &permissions;
     let test_options = &test_options;
-    let ps = ps.clone();
+    ps.borrow_mut().reset_for_file_watcher();
+    let ps = ps.borrow().clone();
 
     async move {
       let specifiers_with_mode = fetch_specifiers_with_test_mode(
@@ -1502,12 +1504,13 @@ pub async fn run_tests_with_watch(
     }
   };
 
+  let clear_screen = !ps.borrow().options.no_clear_screen();
   file_watcher::watch_func(
     resolver,
     operation,
     file_watcher::PrintConfig {
       job_name: "Test".to_string(),
-      clear_screen: !ps.options.no_clear_screen(),
+      clear_screen,
     },
   )
   .await?;
diff --git a/cli/util/file_watcher.rs b/cli/util/file_watcher.rs
index cb8ee419a..05415f2a6 100644
--- a/cli/util/file_watcher.rs
+++ b/cli/util/file_watcher.rs
@@ -322,13 +322,13 @@ where
         continue;
       },
       _ = operation_future => {
+        consume_paths_to_watch(&mut watcher, &mut paths_to_watch_receiver);
         // TODO(bartlomieju): print exit code here?
         info!(
           "{} {} finished. Restarting on file change...",
           colors::intense_blue("Watcher"),
           job_name,
         );
-        consume_paths_to_watch(&mut watcher, &mut paths_to_watch_receiver);
       },
     };
 
diff --git a/test_util/src/lib.rs b/test_util/src/lib.rs
index 0b98f094c..45008cf1f 100644
--- a/test_util/src/lib.rs
+++ b/test_util/src/lib.rs
@@ -971,6 +971,17 @@ async fn main_server(
       );
       Ok(res)
     }
+    (_, "/dynamic_module.ts") => {
+      let mut res = Response::new(Body::from(format!(
+        r#"export const time = {};"#,
+        std::time::SystemTime::now().elapsed().unwrap().as_nanos()
+      )));
+      res.headers_mut().insert(
+        "Content-type",
+        HeaderValue::from_static("application/typescript"),
+      );
+      Ok(res)
+    }
     (_, "/echo_accept") => {
       let accept = req.headers().get("accept").map(|v| v.to_str().unwrap());
       let res = Response::new(Body::from(