Turbopack: Remove turbo_tasks::apply_effects, use Effects::apply instead (#91858)

Having both `apply_effects` and `Effects::apply` is confusing.

`apply_effect`'s API seems a bit bad/dangerous because it combines work that should be done inside of a strongly-consistent operation (collecting the effects and reading their `Vc`s) with work that should be done at the top-level outside of any operation or turbo-task function (applying the effects).

The first commit in this PR removes the `apply_effect` API and tries to improve documentation, including adding a doctest that compiles.

The second commit in this PR is an LLM-driven refactor to remove the `apply_effect` callsites.

Author: bgw

crates/next-api/src/operation.rs

@@ -3,7 +3,7 @@ use bincode::{Decode, Encode};
 use turbo_rcstr::RcStr;
 use turbo_tasks::{
     CollectiblesSource, FxIndexMap, NonLocalValue, OperationValue, OperationVc, ResolvedVc,
-    TaskInput, Vc, debug::ValueDebugFormat, get_effects, trace::TraceRawVcs,
+    TaskInput, Vc, debug::ValueDebugFormat, take_effects, trace::TraceRawVcs,
 };
 use turbopack_core::{diagnostics::Diagnostic, issue::CollectibleIssuesExt};
 
@@ -40,7 +40,7 @@ async fn entrypoints_without_collectibles_operation(
     let _ = entrypoints.resolve().strongly_consistent().await?;
     entrypoints.drop_collectibles::<Box<dyn Diagnostic>>();
     entrypoints.drop_issues();
-    let _ = get_effects(entrypoints).await?;
+    let _ = take_effects(entrypoints).await?;
     Ok(entrypoints.connect())
 }
 

crates/next-build-test/src/lib.rs

@@ -12,7 +12,9 @@ use next_api::{
     route::{Endpoint, EndpointOutputPaths, Route, endpoint_write_to_disk},
 };
 use turbo_rcstr::{RcStr, rcstr};
-use turbo_tasks::{ReadConsistency, ResolvedVc, TransientInstance, TurboTasks, Vc, get_effects};
+use turbo_tasks::{
+    Effects, ReadConsistency, ReadRef, ResolvedVc, TransientInstance, TurboTasks, Vc, take_effects,
+};
 use turbo_tasks_backend::{NoopBackingStorage, TurboTasksBackend};
 use turbo_tasks_malloc::TurboMalloc;
 
@@ -186,21 +188,21 @@ pub async fn render_routes(
                             html_endpoint,
                             data_endpoint: _,
                         } => {
-                            endpoint_write_to_disk_with_effects(*html_endpoint).await?;
+                            endpoint_write_to_disk_with_apply(html_endpoint).await?;
                         }
                         Route::PageApi { endpoint } => {
-                            endpoint_write_to_disk_with_effects(*endpoint).await?;
+                            endpoint_write_to_disk_with_apply(endpoint).await?;
                         }
                         Route::AppPage(routes) => {
                             for route in routes {
-                                endpoint_write_to_disk_with_effects(*route.html_endpoint).await?;
+                                endpoint_write_to_disk_with_apply(route.html_endpoint).await?;
                             }
                         }
                         Route::AppRoute {
                             original_name: _,
                             endpoint,
                         } => {
-                            endpoint_write_to_disk_with_effects(*endpoint).await?;
+                            endpoint_write_to_disk_with_apply(endpoint).await?;
                         }
                         Route::Conflict => {
                             tracing::info!("WARN: conflict {}", name);
@@ -243,21 +245,43 @@ pub async fn render_routes(
     Ok(stream.len())
 }
 
-#[turbo_tasks::function]
-async fn endpoint_write_to_disk_with_effects(
+async fn endpoint_write_to_disk_with_apply(
     endpoint: ResolvedVc<Box<dyn Endpoint>>,
-) -> Result<Vc<EndpointOutputPaths>> {
-    let op = endpoint_write_to_disk_operation(endpoint);
-    let result = op.resolve().strongly_consistent().await?;
-    get_effects(op).await?.apply().await?;
-    Ok(*result)
-}
+) -> Result<ReadRef<EndpointOutputPaths>> {
+    #[turbo_tasks::function(operation)]
+    fn inner_operation(endpoint: ResolvedVc<Box<dyn Endpoint>>) -> Vc<EndpointOutputPaths> {
+        // we must wrap this in an operation so we can get the Effects collectibles
+        endpoint_write_to_disk(*endpoint)
+    }
 
-#[turbo_tasks::function(operation)]
-pub fn endpoint_write_to_disk_operation(
-    endpoint: ResolvedVc<Box<dyn Endpoint>>,
-) -> Vc<EndpointOutputPaths> {
-    endpoint_write_to_disk(*endpoint)
+    #[turbo_tasks::value(serialization = "none")]
+    struct WithEffects {
+        output_paths: ReadRef<EndpointOutputPaths>,
+        effects: Effects,
+    }
+
+    #[turbo_tasks::function(operation)]
+    pub async fn inner_operation_with_effects(
+        endpoint: ResolvedVc<Box<dyn Endpoint>>,
+    ) -> Result<Vc<WithEffects>> {
+        let op = inner_operation(endpoint);
+        let output_paths = op.read_strongly_consistent().await?;
+        let effects = take_effects(op).await?;
+        Ok(WithEffects {
+            output_paths,
+            effects,
+        }
+        .cell())
+    }
+
+    let op = inner_operation_with_effects(endpoint);
+    let WithEffects {
+        output_paths,
+        effects,
+    } = &*op.read_strongly_consistent().await?;
+    effects.apply().await?;
+
+    Ok(output_paths.clone())
 }
 
 async fn hmr(

crates/next-napi-bindings/src/next_api/project.rs

@@ -49,9 +49,11 @@ use turbo_rcstr::{RcStr, rcstr};
 use turbo_tasks::{
     Effects, FxIndexSet, NonLocalValue, OperationValue, OperationVc, PrettyPrintError, ReadRef,
     ResolvedVc, TaskInput, TransientInstance, TryJoinIterExt, TurboTasksApi, TurboTasksCallApi,
-    UpdateInfo, Vc, get_effects,
+    UpdateInfo, Vc, mark_top_level_task,
     message_queue::{CompilationEvent, Severity},
+    take_effects,
     trace::TraceRawVcs,
+    unmark_top_level_task_may_leak_eventually_consistent_state,
 };
 use turbo_tasks_backend::{BackingStorage, db_invalidation::invalidation_reasons};
 use turbo_tasks_fs::{
@@ -1840,7 +1842,7 @@ async fn hmr_update_with_issues_operation(
     let filter = project.issue_filter();
     let issues = get_issues(update_op, filter).await?;
     let diagnostics = get_diagnostics(update_op).await?;
-    let effects = Arc::new(get_effects(update_op).await?);
+    let effects = Arc::new(take_effects(update_op).await?);
     Ok(HmrUpdateWithIssues {
         update,
         issues,
@@ -1874,6 +1876,8 @@ pub fn project_hmr_events(
                 let chunk_name: RcStr = outer_chunk_name.clone();
                 let session = session.clone();
                 async move {
+                    // HACK(bgw): Remove this unmark call
+                    unmark_top_level_task_may_leak_eventually_consistent_state();
                     let project = container.project().to_resolved().await?;
                     let state = project
                         .hmr_version_state(chunk_name.clone(), hmr_target, session)
@@ -1893,7 +1897,11 @@ pub fn project_hmr_events(
                         diagnostics,
                         effects,
                     } = &*update;
+                    // HACK(bgw): Remove this mark call
+                    mark_top_level_task();
                     effects.apply().await?;
+                    // HACK(bgw): Remove this unmark call
+                    unmark_top_level_task_may_leak_eventually_consistent_state();
                     match &**update {
                         Update::Missing | Update::None => {}
                         Update::Total(TotalUpdate { to }) => {
@@ -1975,7 +1983,7 @@ async fn get_hmr_chunk_names_with_issues_operation(
     let filter = issue_filter_from_container(container);
     let issues = get_issues(hmr_chunk_names_op, filter).await?;
     let diagnostics = get_diagnostics(hmr_chunk_names_op).await?;
-    let effects = Arc::new(get_effects(hmr_chunk_names_op).await?);
+    let effects = Arc::new(take_effects(hmr_chunk_names_op).await?);
     Ok(HmrChunkNamesWithIssues {
         chunk_names: hmr_chunk_names,
         issues,

crates/next-napi-bindings/src/next_api/utils.rs

@@ -15,7 +15,7 @@ use rustc_hash::FxHashMap;
 use serde::Serialize;
 use turbo_rcstr::RcStr;
 use turbo_tasks::{
-    Effects, OperationVc, ReadRef, TaskId, TryJoinIterExt, Vc, VcValueType, get_effects,
+    Effects, OperationVc, ReadRef, TaskId, TryJoinIterExt, Vc, VcValueType, take_effects,
 };
 use turbo_tasks_fs::FileContent;
 use turbopack_core::{
@@ -486,7 +486,7 @@ pub async fn strongly_consistent_catch_collectables<R: VcValueType + Send>(
     let result = source_op.read_strongly_consistent().await;
     let issues = get_issues(source_op, filter).await?;
     let diagnostics = get_diagnostics(source_op).await?;
-    let effects = Arc::new(get_effects(source_op).await?);
+    let effects = Arc::new(take_effects(source_op).await?);
 
     let result = if result.is_err() && issues.iter().any(|i| i.severity <= IssueSeverity::Error) {
         None