refactor(turbo-tasks): Make State require OperationValue (#73876)

I started writing this PR by requiring `OperationValue` in `State`:

```diff
  impl<T> State<T> {
-     pub fn new(value: T) -> Self {
+     pub fn new(value: T) -> Self
+    where
+        T: OperationValue,
+    {
```

And then went through changing every compilation error that occurred as a result.

## Hacks

**There are some hacks in a number of places here where we incorrectly convert `ResolvedVc` to `OperationValue`.**

Unfortunately these have to exist because it's impractical to convert everything to `OperationValue`. `OperationValues` are harder to construct as they must be created directly from the return value of an unresolved function call.

## Why does `State<T>` need `T: OperationValue`?

`OperationVc` types require that `.connect()` is called before their value can be read.

As an internally-mutable type, `State` allows `Vc`s to be passed in ways that we cannot track. `.connect()` re-connects the `Vc` to the dependency graph so that strong resolution is possible.

Without this, you might end up with stale results when reading a `Vc` inside of `State`, which can be very hard to debug.

Fundamentally, I think `State` is unsound, but this is a band-aid on it until we can migrate to a better solution (likely effect-style `Collectibles` for `VersionedContentMap`).

crates/next-api/src/app.rs

@@ -904,11 +904,6 @@ impl AppEndpoint {
         Ok(app_entry)
     }
 
-    #[turbo_tasks::function]
-    fn output_assets(self: Vc<Self>) -> Vc<OutputAssets> {
-        self.output().output_assets()
-    }
-
     #[turbo_tasks::function]
     async fn output(self: Vc<Self>) -> Result<Vc<AppEndpointOutput>> {
         let this = self.await?;
@@ -1633,7 +1628,7 @@ async fn create_app_paths_manifest(
 #[turbo_tasks::value_impl]
 impl Endpoint for AppEndpoint {
     #[turbo_tasks::function]
-    async fn write_to_disk(self: Vc<Self>) -> Result<Vc<WrittenEndpoint>> {
+    async fn write_to_disk(self: ResolvedVc<Self>) -> Result<Vc<WrittenEndpoint>> {
         let this = self.await?;
         let page_name = this.page.to_string();
         let span = match this.ty {
@@ -1658,9 +1653,8 @@ impl Endpoint for AppEndpoint {
         };
         async move {
             let output = self.output().await?;
-            // Must use self.output_assets() instead of output.output_assets() to make it a
-            // single operation
-            let output_assets = self.output_assets();
+            let output_assets_op = output_assets_operation(self);
+            let output_assets = output_assets_op.connect();
 
             let node_root = this.app_project.project().node_root();
 
@@ -1669,7 +1663,7 @@ impl Endpoint for AppEndpoint {
             let _ = this
                 .app_project
                 .project()
-                .emit_all_output_assets(Vc::cell(output_assets));
+                .emit_all_output_assets(output_assets_op);
 
             let (server_paths, client_paths) = if this
                 .app_project
@@ -1740,6 +1734,11 @@ impl Endpoint for AppEndpoint {
     }
 }
 
+#[turbo_tasks::function(operation)]
+fn output_assets_operation(endpoint: ResolvedVc<AppEndpoint>) -> Vc<OutputAssets> {
+    endpoint.output().output_assets()
+}
+
 #[turbo_tasks::value]
 enum AppEndpointOutput {
     NodeJs {

crates/next-api/src/instrumentation.rs

@@ -244,15 +244,16 @@ struct InstrumentationCoreModules {
 #[turbo_tasks::value_impl]
 impl Endpoint for InstrumentationEndpoint {
     #[turbo_tasks::function]
-    async fn write_to_disk(self: Vc<Self>) -> Result<Vc<WrittenEndpoint>> {
+    async fn write_to_disk(self: ResolvedVc<Self>) -> Result<Vc<WrittenEndpoint>> {
         let span = tracing::info_span!("instrumentation endpoint");
         async move {
             let this = self.await?;
-            let output_assets = self.output_assets();
+            let output_assets_op = output_assets_operation(self);
+            let output_assets = output_assets_op.connect();
             let _ = output_assets.resolve().await?;
             let _ = this
                 .project
-                .emit_all_output_assets(Vc::cell(output_assets))
+                .emit_all_output_assets(output_assets_op)
                 .resolve()
                 .await?;
 
@@ -294,3 +295,8 @@ impl Endpoint for InstrumentationEndpoint {
         ]))
     }
 }
+
+#[turbo_tasks::function(operation)]
+fn output_assets_operation(endpoint: ResolvedVc<InstrumentationEndpoint>) -> Vc<OutputAssets> {
+    endpoint.output_assets()
+}

crates/next-api/src/middleware.rs

@@ -269,15 +269,16 @@ impl MiddlewareEndpoint {
 #[turbo_tasks::value_impl]
 impl Endpoint for MiddlewareEndpoint {
     #[turbo_tasks::function]
-    async fn write_to_disk(self: Vc<Self>) -> Result<Vc<WrittenEndpoint>> {
+    async fn write_to_disk(self: ResolvedVc<Self>) -> Result<Vc<WrittenEndpoint>> {
         let span = tracing::info_span!("middleware endpoint");
         async move {
             let this = self.await?;
-            let output_assets = self.output_assets();
+            let output_assets_op = output_assets_operation(self);
+            let output_assets = output_assets_op.connect();
             let _ = output_assets.resolve().await?;
             let _ = this
                 .project
-                .emit_all_output_assets(Vc::cell(output_assets))
+                .emit_all_output_assets(output_assets_op)
                 .resolve()
                 .await?;
 
@@ -324,3 +325,8 @@ impl Endpoint for MiddlewareEndpoint {
         Ok(Vc::cell(vec![self.userland_module().to_resolved().await?]))
     }
 }
+
+#[turbo_tasks::function(operation)]
+fn output_assets_operation(endpoint: ResolvedVc<MiddlewareEndpoint>) -> Vc<OutputAssets> {
+    endpoint.output_assets()
+}

crates/next-api/src/pages.rs

@@ -1122,11 +1122,6 @@ impl PageEndpoint {
         ))
     }
 
-    #[turbo_tasks::function]
-    fn output_assets(self: Vc<Self>) -> Vc<OutputAssets> {
-        self.output().output_assets()
-    }
-
     #[turbo_tasks::function]
     async fn output(self: Vc<Self>) -> Result<Vc<PageEndpointOutput>> {
         let this = self.await?;
@@ -1338,7 +1333,7 @@ pub struct InternalSsrChunkModule {
 #[turbo_tasks::value_impl]
 impl Endpoint for PageEndpoint {
     #[turbo_tasks::function]
-    async fn write_to_disk(self: Vc<Self>) -> Result<Vc<WrittenEndpoint>> {
+    async fn write_to_disk(self: ResolvedVc<Self>) -> Result<Vc<WrittenEndpoint>> {
         let this = self.await?;
         let original_name = this.original_name.await?;
         let span = {
@@ -1359,14 +1354,13 @@ impl Endpoint for PageEndpoint {
         };
         async move {
             let output = self.output().await?;
-            // Must use self.output_assets() instead of output.output_assets() to make it a
-            // single operation
-            let output_assets = self.output_assets();
+            let output_assets_op = output_assets_operation(self);
+            let output_assets = output_assets_op.connect();
 
             let _ = this
                 .pages_project
                 .project()
-                .emit_all_output_assets(Vc::cell(output_assets))
+                .emit_all_output_assets(output_assets_op)
                 .resolve()
                 .await?;
 
@@ -1451,6 +1445,11 @@ impl Endpoint for PageEndpoint {
     }
 }
 
+#[turbo_tasks::function(operation)]
+fn output_assets_operation(endpoint: ResolvedVc<PageEndpoint>) -> Vc<OutputAssets> {
+    endpoint.output().output_assets()
+}
+
 #[turbo_tasks::value]
 enum PageEndpointOutput {
     NodeJs {

crates/next-api/src/project.rs

@@ -28,8 +28,8 @@ use turbo_tasks::{
     fxindexmap,
     graph::{AdjacencyMap, GraphTraversal},
     trace::TraceRawVcs,
-    Completion, Completions, FxIndexMap, IntoTraitRef, NonLocalValue, OperationValue, ReadRef,
-    ResolvedVc, State, TaskInput, TransientInstance, TryFlatJoinIterExt, Value, Vc,
+    Completion, Completions, FxIndexMap, IntoTraitRef, NonLocalValue, OperationValue, OperationVc,
+    ReadRef, ResolvedVc, State, TaskInput, TransientInstance, TryFlatJoinIterExt, Value, Vc,
 };
 use turbo_tasks_env::{EnvMap, ProcessEnv};
 use turbo_tasks_fs::{DiskFileSystem, FileSystem, FileSystemPath, VirtualFileSystem};
@@ -69,7 +69,7 @@ use crate::{
     middleware::MiddlewareEndpoint,
     pages::PagesProject,
     route::{AppPageRoute, Endpoint, Route},
-    versioned_content_map::{OutputAssetsOperation, VersionedContentMap},
+    versioned_content_map::VersionedContentMap,
 };
 
 #[derive(
@@ -259,7 +259,7 @@ impl ProjectContainer {
             // we only need to enable versioning in dev mode, since build
             // is assumed to be operating over a static snapshot
             versioned_content_map: if dev {
-                Some(VersionedContentMap::new().to_resolved().await?)
+                Some(VersionedContentMap::new())
             } else {
                 None
             },
@@ -1300,7 +1300,7 @@ impl Project {
     #[turbo_tasks::function]
     pub async fn emit_all_output_assets(
         self: Vc<Self>,
-        output_assets: Vc<OutputAssetsOperation>,
+        output_assets: OperationVc<OutputAssets>,
     ) -> Result<()> {
         let span = tracing::info_span!("emitting");
         async move {
@@ -1323,7 +1323,7 @@ impl Project {
                 Ok(())
             } else {
                 let _ = emit_assets(
-                    *all_output_assets.await?,
+                    all_output_assets.connect(),
                     node_root,
                     client_relative_path,
                     node_root,
@@ -1500,21 +1500,14 @@ async fn get_referenced_output_assets(
     Ok(parent.references().await?.clone_value().into_iter())
 }
 
-#[turbo_tasks::function]
-async fn all_assets_from_entries_operation_inner(
-    operation: Vc<OutputAssetsOperation>,
+#[turbo_tasks::function(operation)]
+async fn all_assets_from_entries_operation(
+    operation: OperationVc<OutputAssets>,
 ) -> Result<Vc<OutputAssets>> {
-    let assets = *operation.await?;
-    Vc::connect(assets);
+    let assets = operation.connect();
     Ok(all_assets_from_entries(assets))
 }
 
-fn all_assets_from_entries_operation(
-    operation: Vc<OutputAssetsOperation>,
-) -> Vc<OutputAssetsOperation> {
-    Vc::cell(all_assets_from_entries_operation_inner(operation))
-}
-
 #[turbo_tasks::function]
 fn stable_endpoint(endpoint: Vc<Box<dyn Endpoint>>) -> Vc<Box<dyn Endpoint>> {
     endpoint