refactor(turbopack): Rewrite CollectiblesSource callsites to use OperationVc (part 2/3)

crates/napi/src/next_api/endpoint.rs

@@ -4,10 +4,15 @@ use anyhow::Result;
 use napi::{bindgen_prelude::External, JsFunction};
 use next_api::{
     paths::ServerPath,
-    route::{Endpoint, WrittenEndpoint},
+    route::{
+        endpoint_server_changed_operation, endpoint_write_to_disk_operation, Endpoint,
+        WrittenEndpoint,
+    },
 };
 use tracing::Instrument;
-use turbo_tasks::{get_effects, Completion, Effects, ReadRef, Vc, VcValueType};
+use turbo_tasks::{
+    get_effects, Completion, Effects, OperationVc, ReadRef, ResolvedVc, Vc, VcValueType,
+};
 use turbopack_core::{
     diagnostics::PlainDiagnostic,
     error::PrettyPrintError,
@@ -99,14 +104,14 @@ impl Deref for ExternalEndpoint {
 // Await the source and return fatal issues if there are any, otherwise
 // propagate any actual error results.
 async fn strongly_consistent_catch_collectables<R: VcValueType + Send>(
-    source: Vc<R>,
+    source: OperationVc<R>,
 ) -> Result<(
     Option<ReadRef<R>>,
     Arc<Vec<ReadRef<PlainIssue>>>,
     Arc<Vec<ReadRef<PlainDiagnostic>>>,
     Arc<Effects>,
 )> {
-    let result = source.strongly_consistent().await;
+    let result = source.connect().strongly_consistent().await;
     let issues = get_issues(source).await?;
     let diagnostics = get_diagnostics(source).await?;
     let effects = Arc::new(get_effects(source).await?);
@@ -130,9 +135,9 @@ struct WrittenEndpointWithIssues {
 
 #[turbo_tasks::function]
 async fn get_written_endpoint_with_issues(
-    endpoint: Vc<Box<dyn Endpoint>>,
+    endpoint: ResolvedVc<Box<dyn Endpoint>>,
 ) -> Result<Vc<WrittenEndpointWithIssues>> {
-    let write_to_disk = endpoint.write_to_disk();
+    let write_to_disk = endpoint_write_to_disk_operation(endpoint);
     let (written, issues, diagnostics, effects) =
         strongly_consistent_catch_collectables(write_to_disk).await?;
     Ok(WrittenEndpointWithIssues {
@@ -236,10 +241,10 @@ impl Eq for EndpointIssuesAndDiags {}
 
 #[turbo_tasks::function]
 async fn subscribe_issues_and_diags(
-    endpoint: Vc<Box<dyn Endpoint>>,
+    endpoint: ResolvedVc<Box<dyn Endpoint>>,
     should_include_issues: bool,
 ) -> Result<Vc<EndpointIssuesAndDiags>> {
-    let changed = endpoint.server_changed();
+    let changed = endpoint_server_changed_operation(endpoint);
 
     if should_include_issues {
         let (changed_value, issues, diagnostics, effects) =
@@ -252,7 +257,7 @@ async fn subscribe_issues_and_diags(
         }
         .cell())
     } else {
-        let changed_value = changed.strongly_consistent().await?;
+        let changed_value = changed.connect().strongly_consistent().await?;
         Ok(EndpointIssuesAndDiags {
             changed: Some(changed_value),
             issues: Arc::new(vec![]),
@@ -276,7 +281,7 @@ pub fn endpoint_client_changed_subscribe(
         move || {
             async move {
                 let changed = endpoint.client_changed();
-                // We don't capture issues and diagonistics here since we don't want to be
+                // We don't capture issues and diagnostics here since we don't want to be
                 // notified when they change
                 changed.strongly_consistent().await?;
                 Ok(())

crates/napi/src/next_api/project.rs

@@ -632,10 +632,13 @@
 
 #[turbo_tasks::function]
 async fn get_entrypoints_with_issues(
-    container: Vc<ProjectContainer>,
+    container: ResolvedVc<ProjectContainer>,
 ) -> Result<Vc<EntrypointsWithIssues>> {
-    let entrypoints_operation = container.entrypoints();
-    let entrypoints = entrypoints_operation.strongly_consistent().await?;
+    let entrypoints_operation = project_container_entrypoints_operation(container);
+    let entrypoints = entrypoints_operation
+        .connect()
+        .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?);
@@ -648,6 +651,13 @@
     .cell())
 }
 
+#[turbo_tasks::function(operation)]
+fn project_container_entrypoints_operation(
+    container: ResolvedVc<ProjectContainer>,
+) -> Vc<Entrypoints> {
+    container.entrypoints()
+}
+
 #[napi(ts_return_type = "{ __napiType: \"RootTask\" }")]
 pub fn project_entrypoints_subscribe(
     #[napi(ts_arg_type = "{ __napiType: \"Project\" }")] project: External<ProjectInstance>,
@@ -731,12 +741,12 @@
 
 #[turbo_tasks::function]
 async fn hmr_update(
-    project: Vc<Project>,
+    project: ResolvedVc<Project>,
     identifier: RcStr,
-    state: Vc<VersionState>,
+    state: ResolvedVc<VersionState>,
 ) -> Result<Vc<HmrUpdateWithIssues>> {
-    let update_operation = project.hmr_update(identifier, state);
-    let update = update_operation.strongly_consistent().await?;
+    let update_operation = project_hmr_update_operation(project, identifier, state);
+    let update = update_operation.connect().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?);
@@ -749,6 +759,15 @@
     .cell())
 }
 
+#[turbo_tasks::function(operation)]
+async fn project_hmr_update_operation(
+    project: ResolvedVc<Project>,
+    identifier: RcStr,
+    state: ResolvedVc<VersionState>,
+) -> Vc<Update> {
+    project.hmr_update(identifier, *state)
+}
+
 #[napi(ts_return_type = "{ __napiType: \"RootTask\" }")]
 pub fn project_hmr_events(
     #[napi(ts_arg_type = "{ __napiType: \"Project\" }")] project: External<ProjectInstance>,
@@ -849,10 +868,13 @@
 
 #[turbo_tasks::function]
 async fn get_hmr_identifiers_with_issues(
-    container: Vc<ProjectContainer>,
+    container: ResolvedVc<ProjectContainer>,
 ) -> Result<Vc<HmrIdentifiersWithIssues>> {
-    let hmr_identifiers_operation = container.hmr_identifiers();
-    let hmr_identifiers = hmr_identifiers_operation.strongly_consistent().await?;
+    let hmr_identifiers_operation = project_container_hmr_identifiers_operation(container);
+    let hmr_identifiers = hmr_identifiers_operation
+        .connect()
+        .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?);
@@ -865,6 +887,13 @@
     .cell())
 }
 
+#[turbo_tasks::function(operation)]
+fn project_container_hmr_identifiers_operation(
+    container: ResolvedVc<ProjectContainer>,
+) -> Vc<Vec<RcStr>> {
+    container.hmr_identifiers()
+}
+
 #[napi(ts_return_type = "{ __napiType: \"RootTask\" }")]
 pub fn project_hmr_identifiers_subscribe(
     #[napi(ts_arg_type = "{ __napiType: \"Project\" }")] project: External<ProjectInstance>,
@@ -951,7 +980,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/napi/src/next_api/utils.rs

@@ -10,7 +10,7 @@ use napi::{
 };
 use serde::Serialize;
 use turbo_tasks::{
-    trace::TraceRawVcs, ReadRef, TaskId, TryJoinIterExt, TurboTasks, UpdateInfo, Vc,
+    trace::TraceRawVcs, OperationVc, ReadRef, TaskId, TryJoinIterExt, TurboTasks, UpdateInfo, Vc,
 };
 use turbo_tasks_backend::{default_backing_storage, DefaultBackingStorage};
 use turbo_tasks_fs::FileContent;
@@ -229,7 +229,7 @@ pub fn root_task_dispose(
     Ok(())
 }
 
-pub async fn get_issues<T: Send>(source: Vc<T>) -> Result<Arc<Vec<ReadRef<PlainIssue>>>> {
+pub async fn get_issues<T: Send>(source: OperationVc<T>) -> Result<Arc<Vec<ReadRef<PlainIssue>>>> {
     let issues = source.peek_issues_with_path().await?;
     Ok(Arc::new(issues.get_plain_issues().await?))
 }
@@ -238,7 +238,9 @@ pub async fn get_issues<T: Send>(source: Vc<T>) -> Result<Arc<Vec<ReadRef<PlainI
 /// by the given source and returns it as a
 /// [turbopack_core::diagnostics::PlainDiagnostic]. It does
 /// not consume any Diagnostics held by the source.
-pub async fn get_diagnostics<T: Send>(source: Vc<T>) -> Result<Arc<Vec<ReadRef<PlainDiagnostic>>>> {
+pub async fn get_diagnostics<T: Send>(
+    source: OperationVc<T>,
+) -> Result<Arc<Vec<ReadRef<PlainDiagnostic>>>> {
     let captured_diags = source.peek_diagnostics().await?;
     let mut diags = captured_diags
         .diagnostics

crates/next-api/src/route.rs

@@ -66,6 +66,20 @@ pub trait Endpoint {
     fn root_modules(self: Vc<Self>) -> Vc<Modules>;
 }
 
+#[turbo_tasks::function(operation)]
+pub fn endpoint_write_to_disk_operation(
+    endpoint: ResolvedVc<Box<dyn Endpoint>>,
+) -> Vc<WrittenEndpoint> {
+    endpoint.write_to_disk()
+}
+
+#[turbo_tasks::function(operation)]
+pub fn endpoint_server_changed_operation(
+    endpoint: ResolvedVc<Box<dyn Endpoint>>,
+) -> Vc<Completion> {
+    endpoint.server_changed()
+}
+
 #[turbo_tasks::value(shared)]
 #[derive(Debug, Clone)]
 pub enum WrittenEndpoint {