fix: ensure concurrent non-statically analyzable dynamic imports do not sometimes fail (#17923)

Closes #17918

cli/graph_util.rs

@@ -3,6 +3,7 @@
 use crate::args::CliOptions;
 use crate::args::Lockfile;
 use crate::args::TsConfigType;
+use crate::args::TsTypeLib;
 use crate::args::TypeCheckMode;
 use crate::cache;
 use crate::cache::TypeCheckCache;
@@ -16,6 +17,7 @@ use crate::tools::check;
 use deno_core::anyhow::bail;
 use deno_core::error::custom_error;
 use deno_core::error::AnyError;
+use deno_core::parking_lot::RwLock;
 use deno_core::ModuleSpecifier;
 use deno_graph::Module;
 use deno_graph::ModuleGraph;
@@ -24,7 +26,11 @@ use deno_graph::ResolutionError;
 use deno_graph::SpecifierError;
 use deno_runtime::permissions::PermissionsContainer;
 use import_map::ImportMapError;
+use std::collections::HashMap;
+use std::collections::HashSet;
 use std::sync::Arc;
+use tokio::sync::Semaphore;
+use tokio::sync::SemaphorePermit;
 
 #[derive(Clone, Copy)]
 pub struct GraphValidOptions {
@@ -309,6 +315,111 @@ fn get_resolution_error_bare_specifier(
   }
 }
 
+#[derive(Default, Debug)]
+struct GraphData {
+  graph: Arc<ModuleGraph>,
+  checked_libs: HashMap<TsTypeLib, HashSet<ModuleSpecifier>>,
+}
+
+/// Holds the `ModuleGraph` and what parts of it are type checked.
+#[derive(Clone)]
+pub struct ModuleGraphContainer {
+  update_semaphore: Arc<Semaphore>,
+  graph_data: Arc<RwLock<GraphData>>,
+}
+
+impl Default for ModuleGraphContainer {
+  fn default() -> Self {
+    Self {
+      update_semaphore: Arc::new(Semaphore::new(1)),
+      graph_data: Default::default(),
+    }
+  }
+}
+
+impl ModuleGraphContainer {
+  /// Acquires a permit to modify the module graph without other code
+  /// having the chance to modify it. In the meantime, other code may
+  /// still read from the existing module graph.
+  pub async fn acquire_update_permit(&self) -> ModuleGraphUpdatePermit {
+    let permit = self.update_semaphore.acquire().await.unwrap();
+    ModuleGraphUpdatePermit {
+      permit,
+      graph_data: self.graph_data.clone(),
+      graph: (*self.graph_data.read().graph).clone(),
+    }
+  }
+
+  pub fn graph(&self) -> Arc<ModuleGraph> {
+    self.graph_data.read().graph.clone()
+  }
+
+  /// Mark `roots` and all of their dependencies as type checked under `lib`.
+  /// Assumes that all of those modules are known.
+  pub fn set_type_checked(&self, roots: &[ModuleSpecifier], lib: TsTypeLib) {
+    // It's ok to analyze and update this while the module graph itself is
+    // being updated in a permit because the module graph update is always
+    // additive and this will be a subset of the original graph
+    let graph = self.graph();
+    let entries = graph.walk(
+      roots,
+      deno_graph::WalkOptions {
+        check_js: true,
+        follow_dynamic: true,
+        follow_type_only: true,
+      },
+    );
+
+    // now update
+    let mut data = self.graph_data.write();
+    let checked_lib_set = data.checked_libs.entry(lib).or_default();
+    for (specifier, _) in entries {
+      checked_lib_set.insert(specifier.clone());
+    }
+  }
+
+  /// Check if `roots` are all marked as type checked under `lib`.
+  pub fn is_type_checked(
+    &self,
+    roots: &[ModuleSpecifier],
+    lib: TsTypeLib,
+  ) -> bool {
+    let data = self.graph_data.read();
+    match data.checked_libs.get(&lib) {
+      Some(checked_lib_set) => roots.iter().all(|r| {
+        let found = data.graph.resolve(r);
+        checked_lib_set.contains(&found)
+      }),
+      None => false,
+    }
+  }
+}
+
+/// A permit for updating the module graph. When complete and
+/// everything looks fine, calling `.commit()` will store the
+/// new graph in the ModuleGraphContainer.
+pub struct ModuleGraphUpdatePermit<'a> {
+  permit: SemaphorePermit<'a>,
+  graph_data: Arc<RwLock<GraphData>>,
+  graph: ModuleGraph,
+}
+
+impl<'a> ModuleGraphUpdatePermit<'a> {
+  /// Gets the module graph for mutation.
+  pub fn graph_mut(&mut self) -> &mut ModuleGraph {
+    &mut self.graph
+  }
+
+  /// Saves the mutated module graph in the container
+  /// and returns an Arc to the new module graph.
+  pub fn commit(self) -> Arc<ModuleGraph> {
+    let graph = Arc::new(self.graph);
+    self.graph_data.write().graph = graph.clone();
+    drop(self.permit); // explicit drop for clarity
+    graph
+  }
+}
+
 #[cfg(test)]
 mod test {
   use std::sync::Arc;

cli/proc_state.rs

@@ -20,6 +20,7 @@ use crate::file_fetcher::FileFetcher;
 use crate::graph_util::build_graph_with_npm_resolution;
 use crate::graph_util::graph_lock_or_exit;
 use crate::graph_util::graph_valid_with_cli_options;
+use crate::graph_util::ModuleGraphContainer;
 use crate::http_util::HttpClient;
 use crate::node;
 use crate::node::NodeResolution;
@@ -38,7 +39,6 @@ use deno_core::error::custom_error;
 use deno_core::error::generic_error;
 use deno_core::error::AnyError;
 use deno_core::parking_lot::Mutex;
-use deno_core::parking_lot::RwLock;
 use deno_core::resolve_url_or_path;
 use deno_core::CompiledWasmModuleStore;
 use deno_core::ModuleSpecifier;
@@ -58,7 +58,6 @@ use deno_runtime::permissions::PermissionsContainer;
 use import_map::ImportMap;
 use log::warn;
 use std::borrow::Cow;
-use std::collections::HashMap;
 use std::collections::HashSet;
 use std::ops::Deref;
 use std::path::PathBuf;
@@ -78,7 +77,7 @@ pub struct Inner {
   pub emit_cache: EmitCache,
   pub emit_options: deno_ast::EmitOptions,
   pub emit_options_hash: u64,
-  graph_data: Arc<RwLock<GraphData>>,
+  graph_container: ModuleGraphContainer,
   pub lockfile: Option<Arc<Mutex<Lockfile>>>,
   pub maybe_import_map: Option<Arc<ImportMap>>,
   pub maybe_inspector_server: Option<Arc<InspectorServer>>,
@@ -139,7 +138,7 @@ impl ProcState {
       emit_options: self.emit_options.clone(),
       file_fetcher: self.file_fetcher.clone(),
       http_client: self.http_client.clone(),
-      graph_data: Default::default(),
+      graph_container: Default::default(),
       lockfile: self.lockfile.clone(),
       maybe_import_map: self.maybe_import_map.clone(),
       maybe_inspector_server: self.maybe_inspector_server.clone(),
@@ -284,7 +283,7 @@ impl ProcState {
       emit_options,
       file_fetcher: Arc::new(file_fetcher),
       http_client,
-      graph_data: Default::default(),
+      graph_container: Default::default(),
       lockfile,
       maybe_import_map,
       maybe_inspector_server,
@@ -340,7 +339,9 @@ impl ProcState {
     let analyzer = self.parsed_source_cache.as_analyzer();
 
     log::debug!("Creating module graph.");
-    let mut graph = self.graph_data.read().graph_inner_clone();
+    let mut graph_update_permit =
+      self.graph_container.acquire_update_permit().await;
+    let graph = graph_update_permit.graph_mut();
 
     // Determine any modules that have already been emitted this session and
     // should be skipped.
@@ -348,7 +349,7 @@ impl ProcState {
       graph.specifiers().map(|(s, _)| s.clone()).collect();
 
     build_graph_with_npm_resolution(
-      &mut graph,
+      graph,
       &self.npm_resolver,
       roots.clone(),
       &mut cache,
@@ -365,15 +366,12 @@ impl ProcState {
 
     // If there is a lockfile, validate the integrity of all the modules.
     if let Some(lockfile) = &self.lockfile {
-      graph_lock_or_exit(&graph, &mut lockfile.lock());
+      graph_lock_or_exit(graph, &mut lockfile.lock());
     }
 
-    let graph = {
-      graph_valid_with_cli_options(&graph, &roots, &self.options)?;
-      let mut graph_data = self.graph_data.write();
-      graph_data.update_graph(Arc::new(graph));
-      graph_data.graph.clone()
-    };
+    graph_valid_with_cli_options(graph, &roots, &self.options)?;
+    // save the graph and get a reference to the new graph
+    let graph = graph_update_permit.commit();
 
     if graph.has_node_specifier
       && self.options.type_check_mode() != TypeCheckMode::None
@@ -388,14 +386,11 @@ impl ProcState {
 
     // type check if necessary
     if self.options.type_check_mode() != TypeCheckMode::None
-      && !self.graph_data.read().is_type_checked(&roots, lib)
+      && !self.graph_container.is_type_checked(&roots, lib)
     {
       log::debug!("Type checking.");
       let maybe_config_specifier = self.options.maybe_config_file_specifier();
-      let graph = {
-        let graph_data = self.graph_data.read();
-        Arc::new(graph_data.graph.segment(&roots))
-      };
+      let graph = Arc::new(graph.segment(&roots));
       let options = check::CheckOptions {
         type_check_mode: self.options.type_check_mode(),
         debug: self.options.log_level() == Some(log::Level::Debug),
@@ -412,7 +407,7 @@ impl ProcState {
         TypeCheckCache::new(&self.dir.type_checking_cache_db_file_path());
       let check_result =
         check::check(graph, &check_cache, &self.npm_resolver, options)?;
-      self.graph_data.write().set_type_checked(&roots, lib);
+      self.graph_container.set_type_checked(&roots, lib);
       if !check_result.diagnostics.is_empty() {
         return Err(anyhow!(check_result.diagnostics));
       }
@@ -492,8 +487,7 @@ impl ProcState {
           });
       }
 
-      let graph_data = self.graph_data.read();
-      let graph = &graph_data.graph;
+      let graph = self.graph_container.graph();
       let maybe_resolved = match graph.get(&referrer) {
         Some(Module::Esm(module)) => {
           module.dependencies.get(specifier).map(|d| &d.maybe_code)
@@ -684,11 +678,7 @@ impl ProcState {
   }
 
   pub fn graph(&self) -> Arc<ModuleGraph> {
-    self.graph_data.read().graph.clone()
-  }
-
-  pub fn has_node_builtin_specifier(&self) -> bool {
-    self.graph_data.read().graph.has_node_specifier
+    self.graph_container.graph()
   }
 }
 
@@ -715,57 +705,3 @@ impl deno_graph::source::Reporter for FileWatcherReporter {
     }
   }
 }
-
-#[derive(Debug, Default)]
-struct GraphData {
-  graph: Arc<ModuleGraph>,
-  checked_libs: HashMap<TsTypeLib, HashSet<ModuleSpecifier>>,
-}
-
-impl GraphData {
-  /// Store data from `graph` into `self`.
-  pub fn update_graph(&mut self, graph: Arc<ModuleGraph>) {
-    self.graph = graph;
-  }
-
-  // todo(dsherret): remove the need for cloning this (maybe if we used an AsyncRefCell)
-  pub fn graph_inner_clone(&self) -> ModuleGraph {
-    (*self.graph).clone()
-  }
-
-  /// Mark `roots` and all of their dependencies as type checked under `lib`.
-  /// Assumes that all of those modules are known.
-  pub fn set_type_checked(
-    &mut self,
-    roots: &[ModuleSpecifier],
-    lib: TsTypeLib,
-  ) {
-    let entries = self.graph.walk(
-      roots,
-      deno_graph::WalkOptions {
-        check_js: true,
-        follow_dynamic: true,
-        follow_type_only: true,
-      },
-    );
-    let checked_lib_set = self.checked_libs.entry(lib).or_default();
-    for (specifier, _) in entries {
-      checked_lib_set.insert(specifier.clone());
-    }
-  }
-
-  /// Check if `roots` are all marked as type checked under `lib`.
-  pub fn is_type_checked(
-    &self,
-    roots: &[ModuleSpecifier],
-    lib: TsTypeLib,
-  ) -> bool {
-    match self.checked_libs.get(&lib) {
-      Some(checked_lib_set) => roots.iter().all(|r| {
-        let found = self.graph.resolve(r);
-        checked_lib_set.contains(&found)
-      }),
-      None => false,
-    }
-  }
-}

cli/tests/integration/run_tests.rs

@@ -545,6 +545,12 @@ itest!(dynamic_import_already_rejected {
   output: "run/dynamic_import_already_rejected/main.out",
 });
 
+itest!(dynamic_import_concurrent_non_statically_analyzable {
+  args: "run --allow-read --allow-net --quiet run/dynamic_import_concurrent_non_statically_analyzable/main.ts",
+  output: "run/dynamic_import_concurrent_non_statically_analyzable/main.out",
+  http_server: true,
+});
+
 itest!(no_check_imports_not_used_as_values {
     args: "run --config run/no_check_imports_not_used_as_values/preserve_imports.tsconfig.json --no-check run/no_check_imports_not_used_as_values/main.ts",
     output: "run/no_check_imports_not_used_as_values/main.out",

cli/tests/testdata/run/dynamic_import_concurrent_non_statically_analyzable/main.out

@@ -0,0 +1,100 @@
+0
+1
+2
+3
+4
+5
+6
+7
+8
+9
+10
+11
+12
+13
+14
+15
+16
+17
+18
+19
+20
+21
+22
+23
+24
+25
+26
+27
+28
+29
+30
+31
+32
+33
+34
+35
+36
+37
+38
+39
+40
+41
+42
+43
+44
+45
+46
+47
+48
+49
+50
+51
+52
+53
+54
+55
+56
+57
+58
+59
+60
+61
+62
+63
+64
+65
+66
+67
+68
+69
+70
+71
+72
+73
+74
+75
+76
+77
+78
+79
+80
+81
+82
+83
+84
+85
+86
+87
+88
+89
+90
+91
+92
+93
+94
+95
+96
+97
+98
+99

cli/tests/testdata/run/dynamic_import_concurrent_non_statically_analyzable/main.ts

@@ -0,0 +1,16 @@
+import * as path from "http://localhost:4545/deno_std/path/mod.ts";
+
+const currentDir = path.dirname(path.fromFileUrl(import.meta.url));
+const url = path.toFileUrl(path.join(currentDir, "./mod.ts"));
+const urls = [];
+
+// this is hard to reproduce, but doing this will help
+for (let i = 0; i < 100; i++) {
+  urls.push(url.toString() + "#" + i);
+}
+
+const results = await Promise.all(urls.map((url) => import(url)));
+
+for (const result of results) {
+  result.outputValue();
+}

cli/tests/testdata/run/dynamic_import_concurrent_non_statically_analyzable/mod.ts

@@ -0,0 +1,7 @@
+// sleep a bit so many concurrent tasks end up
+// attempting to build the graph at the same time
+import "http://localhost:4545/sleep/10";
+
+export function outputValue() {
+  console.log(parseInt(new URL(import.meta.url).hash.slice(1), 10));
+}

cli/worker.rs

@@ -293,8 +293,7 @@ impl CliMainWorker {
     &mut self,
     id: ModuleId,
   ) -> Result<(), AnyError> {
-    if self.ps.npm_resolver.has_packages()
-      || self.ps.has_node_builtin_specifier()
+    if self.ps.npm_resolver.has_packages() || self.ps.graph().has_node_specifier
     {
       self.initialize_main_module_for_node().await?;
     }

test_util/src/lib.rs

@@ -44,6 +44,7 @@ use std::sync::Mutex;
 use std::sync::MutexGuard;
 use std::task::Context;
 use std::task::Poll;
+use std::time::Duration;
 use tokio::io::AsyncWriteExt;
 use tokio::net::TcpListener;
 use tokio::net::TcpStream;
@@ -1058,6 +1059,20 @@ async fn main_server(
             return Ok(file_resp);
           }
         }
+      } else if let Some(suffix) = req.uri().path().strip_prefix("/deno_std/") {
+        let mut file_path = std_path();
+        file_path.push(suffix);
+        if let Ok(file) = tokio::fs::read(&file_path).await {
+          let file_resp = custom_headers(req.uri().path(), file);
+          return Ok(file_resp);
+        }
+      } else if let Some(suffix) = req.uri().path().strip_prefix("/sleep/") {
+        let duration = suffix.parse::<u64>().unwrap();
+        tokio::time::sleep(Duration::from_millis(duration)).await;
+        return Response::builder()
+          .status(StatusCode::OK)
+          .header("content-type", "application/typescript")
+          .body(Body::empty());
       }
 
       Response::builder()