[Turbopack] Revert "refactor filesystem writes to an effect based system"

crates/napi/src/next_api/endpoint.rs

@@ -7,7 +7,7 @@ use next_api::{
     route::{Endpoint, WrittenEndpoint},
 };
 use tracing::Instrument;
-use turbo_tasks::{get_effects, Completion, Effects, ReadRef, Vc, VcValueType};
+use turbo_tasks::{Completion, ReadRef, Vc, VcValueType};
 use turbopack_core::{
     diagnostics::PlainDiagnostic,
     error::PrettyPrintError,
@@ -104,42 +104,38 @@ async fn strongly_consistent_catch_collectables<R: VcValueType + Send>(
     Option<ReadRef<R>>,
     Arc<Vec<ReadRef<PlainIssue>>>,
     Arc<Vec<ReadRef<PlainDiagnostic>>>,
-    Arc<Effects>,
 )> {
     let result = source.strongly_consistent().await;
     let issues = get_issues(source).await?;
     let diagnostics = get_diagnostics(source).await?;
-    let effects = Arc::new(get_effects(source).await?);
 
     let result = if result.is_err() && issues.iter().any(|i| i.severity <= IssueSeverity::Error) {
         None
     } else {
         Some(result?)
     };
 
-    Ok((result, issues, diagnostics, effects))
+    Ok((result, issues, diagnostics))
 }
 
 #[turbo_tasks::value(serialization = "none")]
 struct WrittenEndpointWithIssues {
     written: Option<ReadRef<WrittenEndpoint>>,
     issues: Arc<Vec<ReadRef<PlainIssue>>>,
     diagnostics: Arc<Vec<ReadRef<PlainDiagnostic>>>,
-    effects: Arc<Effects>,
 }
 
 #[turbo_tasks::function]
 async fn get_written_endpoint_with_issues(
     endpoint: Vc<Box<dyn Endpoint>>,
 ) -> Result<Vc<WrittenEndpointWithIssues>> {
     let write_to_disk = endpoint.write_to_disk();
-    let (written, issues, diagnostics, effects) =
+    let (written, issues, diagnostics) =
         strongly_consistent_catch_collectables(write_to_disk).await?;
     Ok(WrittenEndpointWithIssues {
         written,
         issues,
         diagnostics,
-        effects,
     }
     .cell())
 }
@@ -153,15 +149,13 @@ pub async fn endpoint_write_to_disk(
     let endpoint = ***endpoint;
     let (written, issues, diags) = turbo_tasks
         .run_once(async move {
-            let operation = get_written_endpoint_with_issues(endpoint);
             let WrittenEndpointWithIssues {
                 written,
                 issues,
                 diagnostics,
-                effects,
-            } = &*operation.strongly_consistent().await?;
-            effects.apply().await?;
-
+            } = &*get_written_endpoint_with_issues(endpoint)
+                .strongly_consistent()
+                .await?;
             Ok((written.clone(), issues.clone(), diagnostics.clone()))
         })
         .await
@@ -186,10 +180,9 @@ pub fn endpoint_server_changed_subscribe(
         func,
         move || {
             async move {
-                let operation = subscribe_issues_and_diags(endpoint, issues);
-                let result = operation.strongly_consistent().await?;
-                result.effects.apply().await?;
-                Ok(result)
+                subscribe_issues_and_diags(endpoint, issues)
+                    .strongly_consistent()
+                    .await
             }
             .instrument(tracing::info_span!("server changes subscription"))
         },
@@ -198,7 +191,6 @@ pub fn endpoint_server_changed_subscribe(
                 changed: _,
                 issues,
                 diagnostics,
-                effects: _,
             } = &*ctx.value;
 
             Ok(vec![TurbopackResult {
@@ -218,7 +210,6 @@ struct EndpointIssuesAndDiags {
     changed: Option<ReadRef<Completion>>,
     issues: Arc<Vec<ReadRef<PlainIssue>>>,
     diagnostics: Arc<Vec<ReadRef<PlainDiagnostic>>>,
-    effects: Arc<Effects>,
 }
 
 impl PartialEq for EndpointIssuesAndDiags {
@@ -242,13 +233,12 @@ async fn subscribe_issues_and_diags(
     let changed = endpoint.server_changed();
 
     if should_include_issues {
-        let (changed_value, issues, diagnostics, effects) =
+        let (changed_value, issues, diagnostics) =
             strongly_consistent_catch_collectables(changed).await?;
         Ok(EndpointIssuesAndDiags {
             changed: changed_value,
             issues,
             diagnostics,
-            effects,
         }
         .cell())
     } else {
@@ -257,7 +247,6 @@ async fn subscribe_issues_and_diags(
             changed: Some(changed_value),
             issues: Arc::new(vec![]),
             diagnostics: Arc::new(vec![]),
-            effects: Arc::new(Effects::default()),
         }
         .cell())
     }

crates/napi/src/next_api/project.rs

@@ -24,7 +24,7 @@
 use tracing::Instrument;
 use tracing_subscriber::{layer::SubscriberExt, util::SubscriberInitExt, EnvFilter, Registry};
 use turbo_rcstr::RcStr;
-use turbo_tasks::{get_effects, Completion, Effects, ReadRef, TransientInstance, UpdateInfo, Vc};
+use turbo_tasks::{Completion, ReadRef, TransientInstance, UpdateInfo, Vc};
 use turbo_tasks_fs::{
     util::uri_from_file, DiskFileSystem, FileContent, FileSystem, FileSystemPath,
 };
@@ -623,7 +623,6 @@
     entrypoints: ReadRef<Entrypoints>,
     issues: Arc<Vec<ReadRef<PlainIssue>>>,
     diagnostics: Arc<Vec<ReadRef<PlainDiagnostic>>>,
-    effects: Arc<Effects>,
 }
 
 #[turbo_tasks::function]
@@ -634,12 +633,10 @@
     let entrypoints = entrypoints_operation.strongly_consistent().await?;
     let issues = get_issues(entrypoints_operation).await?;
     let diagnostics = get_diagnostics(entrypoints_operation).await?;
-    let effects = Arc::new(get_effects(entrypoints_operation).await?);
     Ok(EntrypointsWithIssues {
         entrypoints,
         issues,
         diagnostics,
-        effects,
     }
     .cell())
 }
@@ -656,14 +653,13 @@
         func,
         move || {
             async move {
-                let operation = get_entrypoints_with_issues(container);
                 let EntrypointsWithIssues {
                     entrypoints,
                     issues,
                     diagnostics,
-                    effects,
-                } = &*operation.strongly_consistent().await?;
-                effects.apply().await?;
+                } = &*get_entrypoints_with_issues(container)
+                    .strongly_consistent()
+                    .await?;
                 Ok((entrypoints.clone(), issues.clone(), diagnostics.clone()))
             }
             .instrument(tracing::info_span!("entrypoints subscription"))
@@ -722,7 +718,6 @@
     update: ReadRef<Update>,
     issues: Arc<Vec<ReadRef<PlainIssue>>>,
     diagnostics: Arc<Vec<ReadRef<PlainDiagnostic>>>,
-    effects: Arc<Effects>,
 }
 
 #[turbo_tasks::function]
@@ -735,12 +730,10 @@
     let update = update_operation.strongly_consistent().await?;
     let issues = get_issues(update_operation).await?;
     let diagnostics = get_diagnostics(update_operation).await?;
-    let effects = Arc::new(get_effects(update_operation).await?);
     Ok(HmrUpdateWithIssues {
         update,
         issues,
         diagnostics,
-        effects,
     }
     .cell())
 }
@@ -767,15 +760,14 @@
                     let project = project.project().resolve().await?;
                     let state = project.hmr_version_state(identifier.clone(), session);
 
-                    let operation = hmr_update(project, identifier.clone(), state);
-                    let update = operation.strongly_consistent().await?;
+                    let update = hmr_update(project, identifier.clone(), state)
+                        .strongly_consistent()
+                        .await?;
                     let HmrUpdateWithIssues {
                         update,
                         issues,
                         diagnostics,
-                        effects,
                     } = &*update;
-                    effects.apply().await?;
                     match &**update {
                         Update::Missing | Update::None => {}
                         Update::Total(TotalUpdate { to }) => {
@@ -840,7 +832,6 @@
     identifiers: ReadRef<Vec<RcStr>>,
     issues: Arc<Vec<ReadRef<PlainIssue>>>,
     diagnostics: Arc<Vec<ReadRef<PlainDiagnostic>>>,
-    effects: Arc<Effects>,
 }
 
 #[turbo_tasks::function]
@@ -851,12 +842,10 @@
     let hmr_identifiers = hmr_identifiers_operation.strongly_consistent().await?;
     let issues = get_issues(hmr_identifiers_operation).await?;
     let diagnostics = get_diagnostics(hmr_identifiers_operation).await?;
-    let effects = Arc::new(get_effects(hmr_identifiers_operation).await?);
     Ok(HmrIdentifiersWithIssues {
         identifiers: hmr_identifiers,
         issues,
         diagnostics,
-        effects,
     }
     .cell())
 }
@@ -872,14 +861,13 @@
         turbo_tasks.clone(),
         func,
         move || async move {
-            let operation = get_hmr_identifiers_with_issues(container);
             let HmrIdentifiersWithIssues {
                 identifiers,
                 issues,
                 diagnostics,
-                effects,
-            } = &*operation.strongly_consistent().await?;
-            effects.apply().await?;
+            } = &*get_hmr_identifiers_with_issues(container)
+                .strongly_consistent()
+                .await?;
 
             Ok((identifiers.clone(), issues.clone(), diagnostics.clone()))
         },
@@ -947,7 +935,7 @@
     }
 }
 
 /// Subscribes to lifecycle events of the compilation.
 ///
 /// Emits an [UpdateMessage::Start] event when any computation starts.
 /// Emits an [UpdateMessage::End] event when there was no computation for the

crates/next-api/src/app.rs

@@ -1633,10 +1633,10 @@ impl Endpoint for AppEndpoint {
 
             let node_root_ref = &node_root.await?;
 
-            let _ = this
-                .app_project
+            this.app_project
                 .project()
-                .emit_all_output_assets(Vc::cell(output_assets));
+                .emit_all_output_assets(Vc::cell(output_assets))
+                .await?;
 
             let (server_paths, client_paths) = if this
                 .app_project

crates/next-api/src/instrumentation.rs

@@ -252,7 +252,9 @@ impl Endpoint for InstrumentationEndpoint {
             let this = self.await?;
             let output_assets = self.output_assets();
             let _ = output_assets.resolve().await?;
-            let _ = this.project.emit_all_output_assets(Vc::cell(output_assets));
+            this.project
+                .emit_all_output_assets(Vc::cell(output_assets))
+                .await?;
 
             let server_paths = if this.project.next_mode().await?.is_development() {
                 let node_root = this.project.node_root();

crates/next-api/src/middleware.rs

@@ -277,7 +277,9 @@ impl Endpoint for MiddlewareEndpoint {
             let this = self.await?;
             let output_assets = self.output_assets();
             let _ = output_assets.resolve().await?;
-            let _ = this.project.emit_all_output_assets(Vc::cell(output_assets));
+            this.project
+                .emit_all_output_assets(Vc::cell(output_assets))
+                .await?;
 
             let (server_paths, client_paths) = if this.project.next_mode().await?.is_development() {
                 let node_root = this.project.node_root();

crates/next-api/src/pages.rs

@@ -1269,10 +1269,10 @@ impl Endpoint for PageEndpoint {
             // single operation
             let output_assets = self.output_assets();
 
-            let _ = this
-                .pages_project
+            this.pages_project
                 .project()
-                .emit_all_output_assets(Vc::cell(output_assets));
+                .emit_all_output_assets(Vc::cell(output_assets))
+                .await?;
 
             let node_root = this.pages_project.project().node_root();
 

crates/next-api/src/project.rs

@@ -1166,7 +1166,7 @@ impl Project {
     pub async fn emit_all_output_assets(
         self: Vc<Self>,
         output_assets: Vc<OutputAssetsOperation>,
-    ) -> Result<()> {
+    ) -> Result<Vc<()>> {
         let span = tracing::info_span!("emitting");
         async move {
             let all_output_assets = all_assets_from_entries_operation(output_assets);
@@ -1175,23 +1175,27 @@ impl Project {
             let node_root = self.node_root();
 
             if let Some(map) = self.await?.versioned_content_map {
-                let _ = map.insert_output_assets(
-                    all_output_assets,
-                    node_root,
-                    client_relative_path,
-                    node_root,
-                );
+                let _ = map
+                    .insert_output_assets(
+                        all_output_assets,
+                        node_root,
+                        client_relative_path,
+                        node_root,
+                    )
+                    .resolve()
+                    .await?;
 
-                Ok(())
+                Ok(Vc::cell(()))
             } else {
                 let _ = emit_assets(
                     *all_output_assets.await?,
                     node_root,
                     client_relative_path,
                     node_root,
-                );
-
-                Ok(())
+                )
+                .resolve()
+                .await?;
+                Ok(Vc::cell(()))
             }
         }
         .instrument(span)