fix(npm): resolve dynamic npm imports individually (#24170)

* https://github.com/denoland/deno_npm/pull/57
* https://github.com/denoland/deno_graph/pull/498

Closes https://github.com/denoland/deno/issues/17802

3 files changed, 117 insertions, 306 deletions

diff --git a/cli/npm/managed/installer.rs b/cli/npm/managed/installer.rs
deleted file mode 100644
index 694e01206..000000000
--- a/cli/npm/managed/installer.rs
+++ /dev/null
@@ -1,123 +0,0 @@
-// Copyright 2018-2024 the Deno authors. All rights reserved. MIT license.
-
-use std::future::Future;
-use std::sync::Arc;
-
-use deno_core::error::AnyError;
-use deno_core::futures::stream::FuturesOrdered;
-use deno_core::futures::StreamExt;
-use deno_npm::registry::NpmRegistryApi;
-use deno_npm::registry::NpmRegistryPackageInfoLoadError;
-use deno_semver::package::PackageReq;
-
-use crate::args::PackageJsonDepsProvider;
-use crate::util::sync::AtomicFlag;
-
-use super::CliNpmRegistryApi;
-use super::NpmResolution;
-
-#[derive(Debug)]
-struct PackageJsonDepsInstallerInner {
-  deps_provider: Arc<PackageJsonDepsProvider>,
-  has_installed_flag: AtomicFlag,
-  npm_registry_api: Arc<CliNpmRegistryApi>,
-  npm_resolution: Arc<NpmResolution>,
-}
-
-impl PackageJsonDepsInstallerInner {
-  pub fn reqs_with_info_futures<'a>(
-    &self,
-    reqs: &'a [&'a PackageReq],
-  ) -> FuturesOrdered<
-    impl Future<
-      Output = Result<
-        (&'a PackageReq, Arc<deno_npm::registry::NpmPackageInfo>),
-        NpmRegistryPackageInfoLoadError,
-      >,
-    >,
-  > {
-    FuturesOrdered::from_iter(reqs.iter().map(|req| {
-      let api = self.npm_registry_api.clone();
-      async move {
-        let info = api.package_info(&req.name).await?;
-        Ok::<_, NpmRegistryPackageInfoLoadError>((*req, info))
-      }
-    }))
-  }
-}
-
-/// Holds and controls installing dependencies from package.json.
-#[derive(Debug, Default)]
-pub struct PackageJsonDepsInstaller(Option<PackageJsonDepsInstallerInner>);
-
-impl PackageJsonDepsInstaller {
-  pub fn new(
-    deps_provider: Arc<PackageJsonDepsProvider>,
-    npm_registry_api: Arc<CliNpmRegistryApi>,
-    npm_resolution: Arc<NpmResolution>,
-  ) -> Self {
-    Self(Some(PackageJsonDepsInstallerInner {
-      deps_provider,
-      has_installed_flag: Default::default(),
-      npm_registry_api,
-      npm_resolution,
-    }))
-  }
-
-  /// Creates an installer that never installs local packages during
-  /// resolution. A top level install will be a no-op.
-  pub fn no_op() -> Self {
-    Self(None)
-  }
-
-  /// Installs the top level dependencies in the package.json file
-  /// without going through and resolving the descendant dependencies yet.
-  pub async fn ensure_top_level_install(&self) -> Result<(), AnyError> {
-    let inner = match &self.0 {
-      Some(inner) => inner,
-      None => return Ok(()),
-    };
-
-    if !inner.has_installed_flag.raise() {
-      return Ok(()); // already installed by something else
-    }
-
-    let package_reqs = inner.deps_provider.reqs().unwrap_or_default();
-
-    // check if something needs resolving before bothering to load all
-    // the package information (which is slow)
-    if package_reqs.iter().all(|req| {
-      inner
-        .npm_resolution
-        .resolve_pkg_id_from_pkg_req(req)
-        .is_ok()
-    }) {
-      log::debug!(
-        "All package.json deps resolvable. Skipping top level install."
-      );
-      return Ok(()); // everything is already resolvable
-    }
-
-    let mut reqs_with_info_futures =
-      inner.reqs_with_info_futures(&package_reqs);
-
-    while let Some(result) = reqs_with_info_futures.next().await {
-      let (req, info) = result?;
-      let result = inner
-        .npm_resolution
-        .resolve_pkg_req_as_pending_with_info(req, &info)
-        .await;
-      if let Err(err) = result {
-        if inner.npm_registry_api.mark_force_reload() {
-          log::debug!("Failed to resolve package. Retrying. Error: {err:#}");
-          // re-initialize
-          reqs_with_info_futures = inner.reqs_with_info_futures(&package_reqs);
-        } else {
-          return Err(err.into());
-        }
-      }
-    }
-
-    Ok(())
-  }
-}
diff --git a/cli/npm/managed/mod.rs b/cli/npm/managed/mod.rs
index d3045a1c9..c086235c3 100644
--- a/cli/npm/managed/mod.rs
+++ b/cli/npm/managed/mod.rs
@@ -11,7 +11,6 @@ use deno_core::anyhow::Context;
 use deno_core::error::AnyError;
 use deno_core::parking_lot::Mutex;
 use deno_core::serde_json;
-use deno_graph::NpmPackageReqsResolution;
 use deno_npm::npm_rc::ResolvedNpmRc;
 use deno_npm::registry::NpmPackageInfo;
 use deno_npm::registry::NpmRegistryApi;
@@ -26,6 +25,7 @@ use deno_runtime::deno_node::NodePermissions;
 use deno_runtime::deno_node::NpmResolver;
 use deno_semver::package::PackageNv;
 use deno_semver::package::PackageReq;
+use resolution::AddPkgReqsResult;
 
 use crate::args::Lockfile;
 use crate::args::NpmProcessState;
@@ -35,9 +35,9 @@ use crate::cache::FastInsecureHasher;
 use crate::http_util::HttpClientProvider;
 use crate::util::fs::canonicalize_path_maybe_not_exists_with_fs;
 use crate::util::progress_bar::ProgressBar;
+use crate::util::sync::AtomicFlag;
 
 use self::cache::NpmCache;
-use self::installer::PackageJsonDepsInstaller;
 use self::registry::CliNpmRegistryApi;
 use self::resolution::NpmResolution;
 use self::resolvers::create_npm_fs_resolver;
@@ -48,7 +48,6 @@ use super::InnerCliNpmResolverRef;
 use super::NpmCacheDir;
 
 mod cache;
-mod installer;
 mod registry;
 mod resolution;
 mod resolvers;
@@ -58,11 +57,6 @@ pub enum CliNpmResolverManagedSnapshotOption {
   Specified(Option<ValidSerializedNpmResolutionSnapshot>),
 }
 
-pub enum CliNpmResolverManagedPackageJsonInstallerOption {
-  ConditionalInstall(Arc<PackageJsonDepsProvider>),
-  NoInstall,
-}
-
 pub struct CliNpmResolverManagedCreateOptions {
   pub snapshot: CliNpmResolverManagedSnapshotOption,
   pub maybe_lockfile: Option<Arc<Mutex<Lockfile>>>,
@@ -73,7 +67,7 @@ pub struct CliNpmResolverManagedCreateOptions {
   pub text_only_progress_bar: crate::util::progress_bar::ProgressBar,
   pub maybe_node_modules_path: Option<PathBuf>,
   pub npm_system_info: NpmSystemInfo,
-  pub package_json_installer: CliNpmResolverManagedPackageJsonInstallerOption,
+  pub package_json_deps_provider: Arc<PackageJsonDepsProvider>,
   pub npmrc: Arc<ResolvedNpmRc>,
 }
 
@@ -98,7 +92,7 @@ pub async fn create_managed_npm_resolver_for_lsp(
       npm_api,
       npm_cache,
       options.npmrc,
-      options.package_json_installer,
+      options.package_json_deps_provider,
       options.text_only_progress_bar,
       options.maybe_node_modules_path,
       options.npm_system_info,
@@ -122,7 +116,7 @@ pub async fn create_managed_npm_resolver(
     npm_api,
     npm_cache,
     options.npmrc,
-    options.package_json_installer,
+    options.package_json_deps_provider,
     options.text_only_progress_bar,
     options.maybe_node_modules_path,
     options.npm_system_info,
@@ -138,7 +132,7 @@ fn create_inner(
   npm_api: Arc<CliNpmRegistryApi>,
   npm_cache: Arc<NpmCache>,
   npm_rc: Arc<ResolvedNpmRc>,
-  package_json_installer: CliNpmResolverManagedPackageJsonInstallerOption,
+  package_json_deps_provider: Arc<PackageJsonDepsProvider>,
   text_only_progress_bar: crate::util::progress_bar::ProgressBar,
   node_modules_dir_path: Option<PathBuf>,
   npm_system_info: NpmSystemInfo,
@@ -165,25 +159,13 @@ fn create_inner(
     node_modules_dir_path,
     npm_system_info.clone(),
   );
-  let package_json_deps_installer = match package_json_installer {
-    CliNpmResolverManagedPackageJsonInstallerOption::ConditionalInstall(
-      provider,
-    ) => Arc::new(PackageJsonDepsInstaller::new(
-      provider,
-      npm_api.clone(),
-      resolution.clone(),
-    )),
-    CliNpmResolverManagedPackageJsonInstallerOption::NoInstall => {
-      Arc::new(PackageJsonDepsInstaller::no_op())
-    }
-  };
   Arc::new(ManagedCliNpmResolver::new(
     fs,
     fs_resolver,
     maybe_lockfile,
     npm_api,
     npm_cache,
-    package_json_deps_installer,
+    package_json_deps_provider,
     resolution,
     tarball_cache,
     text_only_progress_bar,
@@ -271,11 +253,12 @@ pub struct ManagedCliNpmResolver {
   maybe_lockfile: Option<Arc<Mutex<Lockfile>>>,
   npm_api: Arc<CliNpmRegistryApi>,
   npm_cache: Arc<NpmCache>,
-  package_json_deps_installer: Arc<PackageJsonDepsInstaller>,
+  package_json_deps_provider: Arc<PackageJsonDepsProvider>,
   resolution: Arc<NpmResolution>,
   tarball_cache: Arc<TarballCache>,
   text_only_progress_bar: ProgressBar,
   npm_system_info: NpmSystemInfo,
+  top_level_install_flag: AtomicFlag,
 }
 
 impl std::fmt::Debug for ManagedCliNpmResolver {
@@ -294,7 +277,7 @@ impl ManagedCliNpmResolver {
     maybe_lockfile: Option<Arc<Mutex<Lockfile>>>,
     npm_api: Arc<CliNpmRegistryApi>,
     npm_cache: Arc<NpmCache>,
-    package_json_deps_installer: Arc<PackageJsonDepsInstaller>,
+    package_json_deps_provider: Arc<PackageJsonDepsProvider>,
     resolution: Arc<NpmResolution>,
     tarball_cache: Arc<TarballCache>,
     text_only_progress_bar: ProgressBar,
@@ -306,11 +289,12 @@ impl ManagedCliNpmResolver {
       maybe_lockfile,
       npm_api,
       npm_cache,
-      package_json_deps_installer,
+      package_json_deps_provider,
       text_only_progress_bar,
       resolution,
       tarball_cache,
       npm_system_info,
+      top_level_install_flag: Default::default(),
     }
   }
 
@@ -385,14 +369,28 @@ impl ManagedCliNpmResolver {
     &self,
     packages: &[PackageReq],
   ) -> Result<(), AnyError> {
+    let result = self.add_package_reqs_raw(packages).await;
+    result.dependencies_result
+  }
+
+  pub async fn add_package_reqs_raw(
+    &self,
+    packages: &[PackageReq],
+  ) -> AddPkgReqsResult {
     if packages.is_empty() {
-      return Ok(());
+      return AddPkgReqsResult {
+        dependencies_result: Ok(()),
+        results: vec![],
+      };
     }
 
-    self.resolution.add_package_reqs(packages).await?;
-    self.fs_resolver.cache_packages().await?;
+    let mut result = self.resolution.add_package_reqs(packages).await;
+    if result.dependencies_result.is_ok() {
+      result.dependencies_result =
+        self.cache_packages().await.map_err(AnyError::from);
+    }
 
-    Ok(())
+    result
   }
 
   /// Sets package requirements to the resolver, removing old requirements and adding new ones.
@@ -422,51 +420,17 @@ impl ManagedCliNpmResolver {
     &self,
   ) -> Result<(), AnyError> {
     // add and ensure this isn't added to the lockfile
-    let package_reqs = vec![PackageReq::from_str("@types/node").unwrap()];
-    self.resolution.add_package_reqs(&package_reqs).await?;
-    self.cache_packages().await?;
+    self
+      .add_package_reqs(&[PackageReq::from_str("@types/node").unwrap()])
+      .await?;
 
     Ok(())
   }
 
-  pub async fn resolve_pending(&self) -> Result<(), AnyError> {
-    self.resolution.resolve_pending().await?;
-    self.cache_packages().await
-  }
-
   pub async fn cache_packages(&self) -> Result<(), AnyError> {
     self.fs_resolver.cache_packages().await
   }
 
-  /// Resolves package requirements for deno graph.
-  pub async fn resolve_npm_for_deno_graph(
-    &self,
-    reqs_with_pkg_infos: &[(&PackageReq, Arc<NpmPackageInfo>)],
-  ) -> NpmPackageReqsResolution {
-    let results = self
-      .resolution
-      .resolve_pkg_reqs_as_pending_with_info(reqs_with_pkg_infos)
-      .await;
-
-    let mut resolutions = Vec::with_capacity(results.len());
-    for result in results {
-      match result {
-        Ok(nv) => {
-          resolutions.push(Ok(nv));
-        }
-        Err(err) => {
-          if self.npm_api.mark_force_reload() {
-            log::debug!("Restarting npm specifier resolution to check for new registry information. Error: {:#}", err);
-            return NpmPackageReqsResolution::ReloadRegistryInfo;
-          } else {
-            resolutions.push(Err(Arc::new(err.into())));
-          }
-        }
-      }
-    }
-    NpmPackageReqsResolution::Resolutions(resolutions)
-  }
-
   pub fn resolve_pkg_folder_from_deno_module(
     &self,
     nv: &PackageNv,
@@ -485,10 +449,26 @@ impl ManagedCliNpmResolver {
   pub async fn ensure_top_level_package_json_install(
     &self,
   ) -> Result<(), AnyError> {
-    self
-      .package_json_deps_installer
-      .ensure_top_level_install()
-      .await
+    let Some(reqs) = self.package_json_deps_provider.reqs() else {
+      return Ok(());
+    };
+    if !self.top_level_install_flag.raise() {
+      return Ok(()); // already did this
+    }
+    // check if something needs resolving before bothering to load all
+    // the package information (which is slow)
+    if reqs
+      .iter()
+      .all(|req| self.resolution.resolve_pkg_id_from_pkg_req(req).is_ok())
+    {
+      log::debug!(
+        "All package.json deps resolvable. Skipping top level install."
+      );
+      return Ok(()); // everything is already resolvable
+    }
+
+    let reqs = reqs.into_iter().cloned().collect::<Vec<_>>();
+    self.add_package_reqs(&reqs).await
   }
 
   pub async fn cache_package_info(
@@ -582,7 +562,7 @@ impl CliNpmResolver for ManagedCliNpmResolver {
       self.maybe_lockfile.clone(),
       self.npm_api.clone(),
       self.npm_cache.clone(),
-      self.package_json_deps_installer.clone(),
+      self.package_json_deps_provider.clone(),
       npm_resolution,
       self.tarball_cache.clone(),
       self.text_only_progress_bar.clone(),
diff --git a/cli/npm/managed/resolution.rs b/cli/npm/managed/resolution.rs
index 7c2756749..c1d31325d 100644
--- a/cli/npm/managed/resolution.rs
+++ b/cli/npm/managed/resolution.rs
@@ -8,9 +8,7 @@ use deno_core::error::AnyError;
 use deno_core::parking_lot::Mutex;
 use deno_lockfile::NpmPackageDependencyLockfileInfo;
 use deno_lockfile::NpmPackageLockfileInfo;
-use deno_npm::registry::NpmPackageInfo;
 use deno_npm::registry::NpmRegistryApi;
-use deno_npm::resolution::NpmPackageVersionResolutionError;
 use deno_npm::resolution::NpmPackagesPartitioned;
 use deno_npm::resolution::NpmResolutionError;
 use deno_npm::resolution::NpmResolutionSnapshot;
@@ -34,6 +32,16 @@ use crate::util::sync::SyncReadAsyncWriteLock;
 
 use super::CliNpmRegistryApi;
 
+pub struct AddPkgReqsResult {
+  /// Results from adding the individual packages.
+  ///
+  /// The indexes of the results correspond to the indexes of the provided
+  /// package requirements.
+  pub results: Vec<Result<PackageNv, NpmResolutionError>>,
+  /// The final result of resolving and caching all the package requirements.
+  pub dependencies_result: Result<(), AnyError>,
+}
+
 /// Handles updating and storing npm resolution in memory where the underlying
 /// snapshot can be updated concurrently. Additionally handles updating the lockfile
 /// based on changes to the resolution.
@@ -80,19 +88,27 @@ impl NpmResolution {
   pub async fn add_package_reqs(
     &self,
     package_reqs: &[PackageReq],
-  ) -> Result<(), AnyError> {
+  ) -> AddPkgReqsResult {
     // only allow one thread in here at a time
     let snapshot_lock = self.snapshot.acquire().await;
-    let snapshot = add_package_reqs_to_snapshot(
+    let result = add_package_reqs_to_snapshot(
       &self.api,
       package_reqs,
       self.maybe_lockfile.clone(),
       || snapshot_lock.read().clone(),
     )
-    .await?;
-
-    *snapshot_lock.write() = snapshot;
-    Ok(())
+    .await;
+
+    AddPkgReqsResult {
+      results: result.results,
+      dependencies_result: match result.dep_graph_result {
+        Ok(snapshot) => {
+          *snapshot_lock.write() = snapshot;
+          Ok(())
+        }
+        Err(err) => Err(err.into()),
+      },
+    }
   }
 
   pub async fn set_package_reqs(
@@ -121,24 +137,8 @@ impl NpmResolution {
         }
       },
     )
-    .await?;
-
-    *snapshot_lock.write() = snapshot;
-
-    Ok(())
-  }
-
-  pub async fn resolve_pending(&self) -> Result<(), AnyError> {
-    // only allow one thread in here at a time
-    let snapshot_lock = self.snapshot.acquire().await;
-
-    let snapshot = add_package_reqs_to_snapshot(
-      &self.api,
-      &Vec::new(),
-      self.maybe_lockfile.clone(),
-      || snapshot_lock.read().clone(),
-    )
-    .await?;
+    .await
+    .into_result()?;
 
     *snapshot_lock.write() = snapshot;
 
@@ -217,49 +217,6 @@ impl NpmResolution {
       .map(|pkg| pkg.id.clone())
   }
 
-  /// Resolves a package requirement for deno graph. This should only be
-  /// called by deno_graph's NpmResolver or for resolving packages in
-  /// a package.json
-  pub async fn resolve_pkg_req_as_pending_with_info(
-    &self,
-    pkg_req: &PackageReq,
-    pkg_info: &NpmPackageInfo,
-  ) -> Result<PackageNv, NpmPackageVersionResolutionError> {
-    debug_assert_eq!(pkg_req.name, pkg_info.name);
-    // only allow one thread in here at a time
-    let snapshot_lock = self.snapshot.acquire().await;
-
-    let mut snapshot = snapshot_lock.write();
-    let pending_resolver = get_npm_pending_resolver(&self.api);
-    let nv = pending_resolver.resolve_package_req_as_pending(
-      &mut snapshot,
-      pkg_req,
-      pkg_info,
-    )?;
-    Ok(nv)
-  }
-
-  pub async fn resolve_pkg_reqs_as_pending_with_info(
-    &self,
-    reqs_with_pkg_infos: &[(&PackageReq, Arc<NpmPackageInfo>)],
-  ) -> Vec<Result<PackageNv, NpmPackageVersionResolutionError>> {
-    // only allow one thread in here at a time
-    let snapshot_lock = self.snapshot.acquire().await;
-
-    let mut snapshot = snapshot_lock.write();
-    let pending_resolver = get_npm_pending_resolver(&self.api);
-    let mut results = Vec::with_capacity(reqs_with_pkg_infos.len());
-    for (pkg_req, pkg_info) in reqs_with_pkg_infos {
-      debug_assert_eq!(pkg_req.name, pkg_info.name);
-      results.push(pending_resolver.resolve_package_req_as_pending(
-        &mut snapshot,
-        pkg_req,
-        pkg_info,
-      ));
-    }
-    results
-  }
-
   pub fn package_reqs(&self) -> HashMap<PackageReq, PackageNv> {
     self.snapshot.read().package_reqs().clone()
   }
@@ -307,52 +264,50 @@ async fn add_package_reqs_to_snapshot(
   package_reqs: &[PackageReq],
   maybe_lockfile: Option<Arc<Mutex<Lockfile>>>,
   get_new_snapshot: impl Fn() -> NpmResolutionSnapshot,
-) -> Result<NpmResolutionSnapshot, AnyError> {
+) -> deno_npm::resolution::AddPkgReqsResult {
   let snapshot = get_new_snapshot();
-  let snapshot = if !snapshot.has_pending()
-    && package_reqs
-      .iter()
-      .all(|req| snapshot.package_reqs().contains_key(req))
+  if package_reqs
+    .iter()
+    .all(|req| snapshot.package_reqs().contains_key(req))
   {
-    log::debug!(
-      "Snapshot already up to date. Skipping pending npm resolution."
-    );
-    snapshot
-  } else {
-    log::debug!(
-      /* this string is used in tests!! */
-      "Running pending npm resolution."
-    );
-    let pending_resolver = get_npm_pending_resolver(api);
-    let result = pending_resolver
-      .resolve_pending(snapshot, package_reqs)
-      .await;
-    api.clear_memory_cache();
-    match result {
-      Ok(snapshot) => snapshot,
-      Err(NpmResolutionError::Resolution(err)) if api.mark_force_reload() => {
-        log::debug!("{err:#}");
-        log::debug!("npm resolution failed. Trying again...");
-
-        // try again
-        let snapshot = get_new_snapshot();
-        let result = pending_resolver
-          .resolve_pending(snapshot, package_reqs)
-          .await;
-        api.clear_memory_cache();
-        // now surface the result after clearing the cache
-        result?
-      }
-      Err(err) => return Err(err.into()),
+    log::debug!("Snapshot already up to date. Skipping npm resolution.");
+    return deno_npm::resolution::AddPkgReqsResult {
+      results: package_reqs
+        .iter()
+        .map(|req| Ok(snapshot.package_reqs().get(req).unwrap().clone()))
+        .collect(),
+      dep_graph_result: Ok(snapshot),
+    };
+  }
+  log::debug!(
+    /* this string is used in tests */
+    "Running npm resolution."
+  );
+  let pending_resolver = get_npm_pending_resolver(api);
+  let result = pending_resolver.add_pkg_reqs(snapshot, package_reqs).await;
+  api.clear_memory_cache();
+  let result = match &result.dep_graph_result {
+    Err(NpmResolutionError::Resolution(err)) if api.mark_force_reload() => {
+      log::debug!("{err:#}");
+      log::debug!("npm resolution failed. Trying again...");
+
+      // try again
+      let snapshot = get_new_snapshot();
+      let result = pending_resolver.add_pkg_reqs(snapshot, package_reqs).await;
+      api.clear_memory_cache();
+      result
     }
+    _ => result,
   };
 
-  if let Some(lockfile_mutex) = maybe_lockfile {
-    let mut lockfile = lockfile_mutex.lock();
-    populate_lockfile_from_snapshot(&mut lockfile, &snapshot);
+  if let Ok(snapshot) = &result.dep_graph_result {
+    if let Some(lockfile_mutex) = maybe_lockfile {
+      let mut lockfile = lockfile_mutex.lock();
+      populate_lockfile_from_snapshot(&mut lockfile, snapshot);
+    }
   }
 
-  Ok(snapshot)
+  result
 }
 
 fn get_npm_pending_resolver(
@@ -374,7 +329,6 @@ fn populate_lockfile_from_snapshot(
   lockfile: &mut Lockfile,
   snapshot: &NpmResolutionSnapshot,
 ) {
-  assert!(!snapshot.has_pending());
   for (package_req, nv) in snapshot.package_reqs() {
     lockfile.insert_package_specifier(
       format!("npm:{}", package_req),