fix: better handling of npm resolution occurring on workers (#24094)

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

1 files changed, 44 insertions, 48 deletions

diff --git a/cli/npm/managed/cache/tarball.rs b/cli/npm/managed/cache/tarball.rs
index a116ad1cf..e65578ff3 100644
--- a/cli/npm/managed/cache/tarball.rs
+++ b/cli/npm/managed/cache/tarball.rs
@@ -8,8 +8,7 @@ use deno_core::anyhow::bail;
 use deno_core::anyhow::Context;
 use deno_core::error::custom_error;
 use deno_core::error::AnyError;
-use deno_core::futures::future::BoxFuture;
-use deno_core::futures::future::Shared;
+use deno_core::futures::future::LocalBoxFuture;
 use deno_core::futures::FutureExt;
 use deno_core::parking_lot::Mutex;
 use deno_npm::npm_rc::ResolvedNpmRc;
@@ -24,6 +23,7 @@ use crate::util::progress_bar::ProgressBar;
 
 use super::tarball_extract::verify_and_extract_tarball;
 use super::tarball_extract::TarballExtractionMode;
+use super::value_creator::MultiRuntimeAsyncValueCreator;
 use super::NpmCache;
 
 // todo(dsherret): create seams and unit test this
@@ -31,7 +31,7 @@ use super::NpmCache;
 #[derive(Debug, Clone)]
 enum MemoryCacheItem {
   /// The cache item hasn't finished yet.
-  PendingFuture(Shared<BoxFuture<'static, Result<(), Arc<AnyError>>>>),
+  Pending(Arc<MultiRuntimeAsyncValueCreator<()>>),
   /// The result errored.
   Errored(Arc<AnyError>),
   /// This package has already been cached.
@@ -71,7 +71,7 @@ impl TarballCache {
   }
 
   pub async fn ensure_package(
-    &self,
+    self: &Arc<Self>,
     package: &PackageNv,
     dist: &NpmPackageVersionDistInfo,
   ) -> Result<(), AnyError> {
@@ -82,69 +82,67 @@ impl TarballCache {
   }
 
   async fn ensure_package_inner(
-    &self,
+    self: &Arc<Self>,
     package_nv: &PackageNv,
     dist: &NpmPackageVersionDistInfo,
   ) -> Result<(), AnyError> {
-    let (created, cache_item) = {
+    let cache_item = {
       let mut mem_cache = self.memory_cache.lock();
       if let Some(cache_item) = mem_cache.get(package_nv) {
-        (false, cache_item.clone())
+        cache_item.clone()
       } else {
         let future = self.create_setup_future(package_nv.clone(), dist.clone());
-        let cache_item = MemoryCacheItem::PendingFuture(future);
+        let value_creator = MultiRuntimeAsyncValueCreator::new(future);
+        let cache_item = MemoryCacheItem::Pending(Arc::new(value_creator));
         mem_cache.insert(package_nv.clone(), cache_item.clone());
-        (true, cache_item)
+        cache_item
       }
     };
 
     match cache_item {
       MemoryCacheItem::Cached => Ok(()),
       MemoryCacheItem::Errored(err) => Err(anyhow!("{}", err)),
-      MemoryCacheItem::PendingFuture(future) => {
-        if created {
-          match future.await {
-            Ok(_) => {
-              *self.memory_cache.lock().get_mut(package_nv).unwrap() =
-                MemoryCacheItem::Cached;
-              Ok(())
-            }
-            Err(err) => {
-              let result_err = anyhow!("{}", err);
-              *self.memory_cache.lock().get_mut(package_nv).unwrap() =
-                MemoryCacheItem::Errored(err);
-              Err(result_err)
-            }
+      MemoryCacheItem::Pending(creator) => {
+        let tarball_cache = self.clone();
+        let result = creator
+          .get(move || {
+            tarball_cache.create_setup_future(package_nv.clone(), dist.clone())
+          })
+          .await;
+        match result {
+          Ok(_) => {
+            *self.memory_cache.lock().get_mut(package_nv).unwrap() =
+              MemoryCacheItem::Cached;
+            Ok(())
+          }
+          Err(err) => {
+            let result_err = anyhow!("{}", err);
+            *self.memory_cache.lock().get_mut(package_nv).unwrap() =
+              MemoryCacheItem::Errored(err);
+            Err(result_err)
           }
-        } else {
-          future.await.map_err(|err| anyhow!("{}", err))
         }
       }
     }
   }
 
   fn create_setup_future(
-    &self,
+    self: &Arc<Self>,
     package_nv: PackageNv,
     dist: NpmPackageVersionDistInfo,
-  ) -> Shared<BoxFuture<'static, Result<(), Arc<AnyError>>>> {
-    let registry_url = self.npmrc.get_registry_url(&package_nv.name);
-    let registry_config =
-      self.npmrc.get_registry_config(&package_nv.name).clone();
-
-    let cache = self.cache.clone();
-    let fs = self.fs.clone();
-    let progress_bar = self.progress_bar.clone();
-    let package_folder =
-      cache.package_folder_for_nv_and_url(&package_nv, registry_url);
-    let http_client_provider = self.http_client_provider.clone();
-
-    deno_core::unsync::spawn(async move {
-      let should_use_cache = cache.should_use_cache_for_package(&package_nv);
-      let package_folder_exists = fs.exists_sync(&package_folder);
+  ) -> LocalBoxFuture<'static, Result<(), AnyError>> {
+    let tarball_cache = self.clone();
+    async move {
+      let registry_url = tarball_cache.npmrc.get_registry_url(&package_nv.name);
+      let registry_config =
+        tarball_cache.npmrc.get_registry_config(&package_nv.name).clone();
+      let package_folder =
+        tarball_cache.cache.package_folder_for_nv_and_url(&package_nv, registry_url);
+      let should_use_cache = tarball_cache.cache.should_use_cache_for_package(&package_nv);
+      let package_folder_exists = tarball_cache.fs.exists_sync(&package_folder);
       if should_use_cache && package_folder_exists {
         return Ok(());
-      } else if cache.cache_setting() == &CacheSetting::Only {
+      } else if tarball_cache.cache.cache_setting() == &CacheSetting::Only {
         return Err(custom_error(
           "NotCached",
           format!(
@@ -162,8 +160,9 @@ impl TarballCache {
       let maybe_auth_header =
         maybe_auth_header_for_npm_registry(&registry_config);
 
-      let guard = progress_bar.update(&dist.tarball);
-      let maybe_bytes = http_client_provider.get_or_create()?
+      let guard = tarball_cache.progress_bar.update(&dist.tarball);
+      let maybe_bytes = tarball_cache.http_client_provider
+        .get_or_create()?
         .download_with_progress(&dist.tarball, maybe_auth_header, &guard)
         .await?;
       match maybe_bytes {
@@ -198,9 +197,6 @@ impl TarballCache {
           bail!("Could not find npm package tarball at: {}", dist.tarball);
         }
       }
-    })
-      .map(|result| result.unwrap().map_err(Arc::new))
-      .boxed()
-      .shared()
+    }.boxed_local()
   }
 }