Fixes performance problems due to TaskScopes (#55721)

### What?

see vercel/turborepo#5992

### Turobopack changes

* vercel/turborepo#6009 <!-- OJ Kwon - ci(workflow):
update test filter -->
* vercel/turborepo#6026 <!-- Will Binns-Smith -
Remove next-dev references and benchmarks -->
* vercel/turborepo#6038 <!-- Tim Neutkens - Remove
test-prod action -->
* vercel/turborepo#6039 <!-- Tim Neutkens - Fix
action dependency -->
* ~vercel/turborepo#6036 <!-- Will Binns-Smith -
Turbopack: add support for an asset prefix (and basePath in Next.js) -->
* vercel/turborepo#5992 <!-- Tobias Koppers -
refactor TaskScopes to use an aggregation tree -->


Closes WEB-1622

Author: sokra

Cargo.lock

@@ -39,11 +39,11 @@ swc_core = { version = "0.83.12", features = [
 testing = { version = "0.34.1" }
 
 # Turbo crates
-turbopack-binding = { git = "https://github.com/vercel/turbo.git", tag = "turbopack-230922.3" }
+turbopack-binding = { git = "https://github.com/vercel/turbo.git", tag = "turbopack-230928.3" }
 # [TODO]: need to refactor embed_directory! macro usages, as well as resolving turbo_tasks::function, macros..
-turbo-tasks = { git = "https://github.com/vercel/turbo.git", tag = "turbopack-230922.3" }
+turbo-tasks = { git = "https://github.com/vercel/turbo.git", tag = "turbopack-230928.3" }
 # [TODO]: need to refactor embed_directory! macro usage in next-core
-turbo-tasks-fs = { git = "https://github.com/vercel/turbo.git", tag = "turbopack-230922.3" }
+turbo-tasks-fs = { git = "https://github.com/vercel/turbo.git", tag = "turbopack-230928.3" }
 
 # General Deps
 
@@ -121,7 +121,7 @@ shadow-rs = { version = "0.23.0", default-features = false, features = [
 ] }
 syn = "1.0.107"
 tempfile = "3.3.0"
-thiserror = "1.0.38"
+thiserror = "1.0.48"
 tiny-gradient = "0.1.0"
 tokio = "1.25.0"
 tokio-util = { version = "0.7.7", features = ["io"] }

Cargo.toml

@@ -77,9 +77,9 @@ pub async fn endpoint_write_to_disk(
     let (written, issues, diags) = turbo_tasks
         .run_once(async move {
             let write_to_disk = endpoint.write_to_disk();
+            let written = write_to_disk.strongly_consistent().await?;
             let issues = get_issues(write_to_disk).await?;
             let diags = get_diagnostics(write_to_disk).await?;
-            let written = write_to_disk.strongly_consistent().await?;
             Ok((written, issues, diags))
         })
         .await
@@ -104,17 +104,16 @@ pub fn endpoint_server_changed_subscribe(
         func,
         move || async move {
             let changed = endpoint.server_changed();
-            let issues = get_issues(changed).await?;
-            let diags = get_diagnostics(changed).await?;
+            // We don't capture issues and diagonistics here since we don't want to be
+            // notified when they change
             changed.strongly_consistent().await?;
-            Ok((issues, diags))
+            Ok(())
         },
-        |ctx| {
-            let (issues, diags) = ctx.value;
+        |_| {
             Ok(vec![TurbopackResult {
                 result: (),
-                issues: issues.iter().map(|i| NapiIssue::from(&**i)).collect(),
-                diagnostics: diags.iter().map(|d| NapiDiagnostic::from(d)).collect(),
+                issues: vec![],
+                diagnostics: vec![],
             }])
         },
     )
@@ -132,17 +131,16 @@ pub fn endpoint_client_changed_subscribe(
         func,
         move || async move {
             let changed = endpoint.client_changed();
-            let issues = get_issues(changed).await?;
-            let diags = get_diagnostics(changed).await?;
+            // We don't capture issues and diagonistics here since we don't want to be
+            // notified when they change
             changed.strongly_consistent().await?;
-            Ok((issues, diags))
+            Ok(())
         },
-        |ctx| {
-            let (issues, diags) = ctx.value;
+        |_| {
             Ok(vec![TurbopackResult {
                 result: (),
-                issues: issues.iter().map(|i| NapiIssue::from(&**i)).collect(),
-                diagnostics: diags.iter().map(|d| NapiDiagnostic::from(d)).collect(),
+                issues: vec![],
+                diagnostics: vec![],
             }])
         },
     )

packages/next-swc/crates/napi/src/next_api/endpoint.rs

@@ -313,11 +313,11 @@ pub fn project_entrypoints_subscribe(
         turbo_tasks.clone(),
         func,
         move || async move {
-            let entrypoints = container.entrypoints();
-            let issues = get_issues(entrypoints).await?;
-            let diags = get_diagnostics(entrypoints).await?;
+            let entrypoints_operation = container.entrypoints();
+            let entrypoints = entrypoints_operation.strongly_consistent().await?;
 
-            let entrypoints = entrypoints.strongly_consistent().await?;
+            let issues = get_issues(entrypoints_operation).await?;
+            let diags = get_diagnostics(entrypoints_operation).await?;
 
             Ok((entrypoints, issues, diags))
         },
@@ -383,10 +383,10 @@ pub fn project_hmr_events(
                     let state = project
                         .project()
                         .hmr_version_state(identifier.clone(), session);
-                    let update = project.project().hmr_update(identifier, state);
-                    let issues = get_issues(update).await?;
-                    let diags = get_diagnostics(update).await?;
-                    let update = update.strongly_consistent().await?;
+                    let update_operation = project.project().hmr_update(identifier, state);
+                    let update = update_operation.strongly_consistent().await?;
+                    let issues = get_issues(update_operation).await?;
+                    let diags = get_diagnostics(update_operation).await?;
                     match &*update {
                         Update::None => {}
                         Update::Total(TotalUpdate { to }) => {
@@ -451,11 +451,11 @@ pub fn project_hmr_identifiers_subscribe(
         turbo_tasks.clone(),
         func,
         move || async move {
-            let hmr_identifiers = container.hmr_identifiers();
-            let issues = get_issues(hmr_identifiers).await?;
-            let diags = get_diagnostics(hmr_identifiers).await?;
+            let hmr_identifiers_operation = container.hmr_identifiers();
+            let hmr_identifiers = hmr_identifiers_operation.strongly_consistent().await?;
 
-            let hmr_identifiers = hmr_identifiers.strongly_consistent().await?;
+            let issues = get_issues(hmr_identifiers_operation).await?;
+            let diags = get_diagnostics(hmr_identifiers_operation).await?;
 
             Ok((hmr_identifiers, issues, diags))
         },

packages/next-swc/crates/napi/src/next_api/project.rs

@@ -69,30 +69,23 @@ impl Drop for RootTask {
 
 #[napi]
 pub fn root_task_dispose(
-    #[napi(ts_arg_type = "{ __napiType: \"RootTask\" }")] _root_task: External<RootTask>,
+    #[napi(ts_arg_type = "{ __napiType: \"RootTask\" }")] mut root_task: External<RootTask>,
 ) -> napi::Result<()> {
-    // TODO(alexkirsz) Implement. Not panicking here to avoid crashing the process
-    // when testing.
+    if let Some(task) = root_task.task_id.take() {
+        root_task.turbo_tasks.dispose_root_task(task);
+    }
     Ok(())
 }
 
 pub async fn get_issues<T: Send>(source: Vc<T>) -> Result<Vec<ReadRef<PlainIssue>>> {
-    let issues = source
-        .peek_issues_with_path()
-        .await?
-        .strongly_consistent()
-        .await?;
+    let issues = source.peek_issues_with_path().await?;
     issues.get_plain_issues().await
 }
 
 /// Collect [turbopack::core::diagnostics::Diagnostic] from given source,
 /// returns [turbopack::core::diagnostics::PlainDiagnostic]
 pub async fn get_diagnostics<T: Send>(source: Vc<T>) -> Result<Vec<ReadRef<PlainDiagnostic>>> {
-    let captured_diags = source
-        .peek_diagnostics()
-        .await?
-        .strongly_consistent()
-        .await?;
+    let captured_diags = source.peek_diagnostics().await?;
 
     captured_diags
         .diagnostics

packages/next-swc/crates/napi/src/next_api/utils.rs

@@ -22,8 +22,8 @@ use turbo_tasks::{
     debug::ValueDebugFormat,
     graph::{AdjacencyMap, GraphTraversal},
     trace::TraceRawVcs,
-    Completion, Completions, IntoTraitRef, State, TaskInput, TransientInstance, TryFlatJoinIterExt,
-    Value, ValueToString, Vc,
+    Completion, Completions, IntoTraitRef, State, TaskInput, TraitRef, TransientInstance,
+    TryFlatJoinIterExt, Value, ValueToString, Vc,
 };
 use turbopack_binding::{
     turbo::{
@@ -562,14 +562,33 @@ impl Project {
             }
         }
 
+        let pages_document_endpoint = TraitRef::cell(
+            self.pages_project()
+                .document_endpoint()
+                .into_trait_ref()
+                .await?,
+        );
+        let pages_app_endpoint =
+            TraitRef::cell(self.pages_project().app_endpoint().into_trait_ref().await?);
+        let pages_error_endpoint = TraitRef::cell(
+            self.pages_project()
+                .error_endpoint()
+                .into_trait_ref()
+                .await?,
+        );
+
         let middleware = find_context_file(
             self.project_path(),
             middleware_files(self.next_config().page_extensions()),
         );
         let middleware = if let FindContextFileResult::Found(fs_path, _) = *middleware.await? {
             let source = Vc::upcast(FileSource::new(fs_path));
             Some(Middleware {
-                endpoint: Vc::upcast(self.middleware_endpoint(source)),
+                endpoint: TraitRef::cell(
+                    Vc::upcast::<Box<dyn Endpoint>>(self.middleware_endpoint(source))
+                        .into_trait_ref()
+                        .await?,
+                ),
             })
         } else {
             None
@@ -578,9 +597,9 @@ impl Project {
         Ok(Entrypoints {
             routes,
             middleware,
-            pages_document_endpoint: self.pages_project().document_endpoint(),
-            pages_app_endpoint: self.pages_project().app_endpoint(),
-            pages_error_endpoint: self.pages_project().error_endpoint(),
+            pages_document_endpoint,
+            pages_app_endpoint,
+            pages_error_endpoint,
         }
         .cell())
     }

packages/next-swc/crates/next-api/src/project.rs

@@ -2,7 +2,11 @@ use std::collections::HashMap;
 
 use anyhow::{bail, Result};
 use next_core::emit_client_assets;
-use turbo_tasks::{State, TryFlatJoinIterExt, TryJoinIterExt, ValueDefault, ValueToString, Vc};
+use serde::{Deserialize, Serialize};
+use turbo_tasks::{
+    debug::ValueDebugFormat, trace::TraceRawVcs, State, TryFlatJoinIterExt, TryJoinIterExt,
+    ValueDefault, ValueToString, Vc,
+};
 use turbopack_binding::{
     turbo::tasks_fs::FileSystemPath,
     turbopack::core::{
@@ -18,8 +22,17 @@ use turbopack_binding::{
 #[turbo_tasks::value(transparent)]
 pub struct OutputAssetsOperation(Vc<OutputAssets>);
 
-type VersionedContentMapInner =
-    HashMap<Vc<FileSystemPath>, (Vc<Box<dyn VersionedContent>>, Vc<OutputAssets>)>;
+#[derive(
+    Clone, Copy, TraceRawVcs, PartialEq, Eq, ValueDebugFormat, Serialize, Deserialize, Debug,
+)]
+struct MapEntry {
+    assets_operation: Vc<OutputAssets>,
+}
+
+#[turbo_tasks::value(transparent)]
+struct OptionMapEntry(Option<MapEntry>);
+
+type VersionedContentMapInner = HashMap<Vc<FileSystemPath>, MapEntry>;
 
 #[turbo_tasks::value]
 pub struct VersionedContentMap {
@@ -55,12 +68,9 @@ impl VersionedContentMap {
         let entries: Vec<_> = assets
             .iter()
             .map(|&asset| async move {
-                // NOTE(alexkirsz) `.versioned_content()` should not be resolved, to ensure that
-                // it always points to the task that computes the versioned
-                // content.
                 Ok((
                     asset.ident().path().resolve().await?,
-                    (asset.versioned_content(), assets_operation),
+                    MapEntry { assets_operation },
                 ))
             })
             .try_join()
@@ -73,14 +83,17 @@ impl VersionedContentMap {
     }
 
     #[turbo_tasks::function]
-    pub async fn get(&self, path: Vc<FileSystemPath>) -> Result<Vc<Box<dyn VersionedContent>>> {
+    pub async fn get(
+        self: Vc<Self>,
+        path: Vc<FileSystemPath>,
+    ) -> Result<Vc<Box<dyn VersionedContent>>> {
         let (content, _) = self.get_internal(path).await?;
         Ok(content)
     }
 
     #[turbo_tasks::function]
     pub async fn get_and_write(
-        &self,
+        self: Vc<Self>,
         path: Vc<FileSystemPath>,
         client_relative_path: Vc<FileSystemPath>,
         client_output_path: Vc<FileSystemPath>,
@@ -105,28 +118,34 @@ impl VersionedContentMap {
             .await?;
         Ok(Vc::cell(keys))
     }
+
+    #[turbo_tasks::function]
+    async fn raw_get(&self, path: Vc<FileSystemPath>) -> Result<Vc<OptionMapEntry>> {
+        let result = {
+            let map = self.map.get();
+            map.get(&path).copied()
+        };
+        Ok(Vc::cell(result))
+    }
 }
 
 impl VersionedContentMap {
     async fn get_internal(
-        &self,
+        self: Vc<Self>,
         path: Vc<FileSystemPath>,
     ) -> Result<(Vc<Box<dyn VersionedContent>>, Vc<OutputAssets>)> {
-        let result = {
-            // NOTE(alexkirsz) This is to avoid Rust marking this method as !Send because a
-            // StateRef to the map is captured across an await boundary below, even though
-            // it does not look like it would.
-            // I think this is a similar issue as https://fasterthanli.me/articles/a-rust-match-made-in-hell
-            let map = self.map.get();
-            map.get(&path).copied()
-        };
-        let Some((content, assets_operation)) = result else {
-            let path = path.to_string().await?;
-            bail!("could not find versioned content for path {}", path);
-        };
-        // NOTE(alexkirsz) This is necessary to mark the task as active again.
-        Vc::connect(assets_operation);
-        Vc::connect(content);
-        Ok((content, assets_operation))
+        let result = self.raw_get(path).await?;
+        if let Some(MapEntry { assets_operation }) = *result {
+            // NOTE(alexkirsz) This is necessary to mark the task as active again.
+            Vc::connect(assets_operation);
+            for asset in assets_operation.await?.iter() {
+                if asset.ident().path().resolve().await? == path {
+                    let content = asset.versioned_content();
+                    return Ok((content, assets_operation));
+                }
+            }
+        }
+        let path = path.to_string().await?;
+        bail!("could not find versioned content for path {}", path);
     }
 }

packages/next-swc/crates/next-api/src/versioned_content_map.rs

@@ -193,7 +193,7 @@
     "@types/ws": "8.2.0",
     "@vercel/ncc": "0.34.0",
     "@vercel/nft": "0.22.6",
-    "@vercel/turbopack-ecmascript-runtime": "https://gitpkg-fork.vercel.sh/vercel/turbo/crates/turbopack-ecmascript-runtime/js?turbopack-230922.2",
+    "@vercel/turbopack-ecmascript-runtime": "https://gitpkg-fork.vercel.sh/vercel/turbo/crates/turbopack-ecmascript-runtime/js?turbopack-230928.3",
     "acorn": "8.5.0",
     "ajv": "8.11.0",
     "amphtml-validator": "1.0.35",

packages/next/package.json

@@ -122,6 +122,8 @@ function handleSuccess() {
     if (isHotUpdate) {
       tryApplyUpdates(onBeforeFastRefresh, onFastRefresh)
     }
+  } else {
+    onBuildOk()
   }
 }
 

packages/next/src/client/dev/error-overlay/hot-dev-client.ts

@@ -1,10 +1,13 @@
 import connect from './error-overlay/hot-dev-client'
 import { sendMessage } from './error-overlay/websocket'
 
+let reloading = false
+
 export default (mode: 'webpack' | 'turbopack') => {
   const devClient = connect(mode)
 
   devClient.subscribeToHmrEvent((obj: any) => {
+    if (reloading) return
     // if we're on an error/404 page, we can't reliably tell if the newly added/removed page
     // matches the current path. In that case, assume any added/removed entries should trigger a reload of the current page
     const isOnErrorPage =
@@ -19,6 +22,7 @@ export default (mode: 'webpack' | 'turbopack') => {
             clientId: window.__nextDevClientId,
           })
         )
+        reloading = true
         return window.location.reload()
       }
       case 'removedPage': {