fix(turbo-tasks): Store persistence of wrapped task on RawVc::LocalOutput

turbopack/crates/turbo-tasks-backend/src/backend/operation/update_output.rs

@@ -91,8 +91,8 @@ impl UpdateOutputOperation {
                     cell,
                 })
             }
-            Ok(Ok(RawVc::LocalOutput(_, _))) => {
-                panic!("LocalOutput must not be output of a task");
+            Ok(Ok(RawVc::LocalOutput(..))) => {
+                panic!("Non-local tasks must not return a local Vc");
             }
             Ok(Err(err)) => {
                 task.insert(CachedDataItem::Error {

turbopack/crates/turbo-tasks-backend/tests/transient_vc.rs

@@ -0,0 +1 @@
+../../turbo-tasks-testing/tests/transient_vc.rs

turbopack/crates/turbo-tasks-memory/tests/transient_vc.rs

@@ -0,0 +1 @@
+../../turbo-tasks-testing/tests/transient_vc.rs

turbopack/crates/turbo-tasks-testing/tests/transient_vc.rs

@@ -0,0 +1,55 @@
+#![feature(arbitrary_self_types)]
+#![feature(arbitrary_self_types_pointers)]
+
+use anyhow::Result;
+use turbo_tasks::{TaskInput, TransientValue, Vc};
+use turbo_tasks_testing::{register, run_without_cache_check, Registration};
+
+static REGISTRATION: Registration = register!();
+
+#[tokio::test]
+async fn test_transient_vc() -> Result<()> {
+    run_without_cache_check(&REGISTRATION, async {
+        test_transient_operation(TransientValue::new(123))
+            .read_strongly_consistent()
+            .await?;
+        Ok(())
+    })
+    .await
+}
+
+#[turbo_tasks::function(operation)]
+async fn test_transient_operation(transient_arg: TransientValue<i32>) -> Result<()> {
+    let called_with_transient = has_transient_arg(transient_arg);
+    let called_with_persistent = has_persistent_arg(123);
+
+    assert!(called_with_transient.is_transient());
+    assert!(!called_with_persistent.is_transient());
+    assert!(has_vc_arg(called_with_transient).is_transient());
+    assert!(!has_vc_arg(called_with_persistent).is_transient());
+
+    let called_with_transient_resolved = called_with_transient.to_resolved().await?;
+    let called_with_persistent_resolved = called_with_persistent.to_resolved().await?;
+
+    assert!(called_with_transient_resolved.is_transient());
+    assert!(!called_with_persistent_resolved.is_transient());
+    assert!(has_vc_arg(*called_with_transient_resolved).is_transient());
+    assert!(!has_vc_arg(*called_with_persistent_resolved).is_transient());
+
+    Ok(())
+}
+
+#[turbo_tasks::function]
+fn has_transient_arg(arg: TransientValue<i32>) -> Vc<i32> {
+    Vc::cell(*arg)
+}
+
+#[turbo_tasks::function]
+fn has_persistent_arg(arg: i32) -> Vc<i32> {
+    Vc::cell(arg)
+}
+
+#[turbo_tasks::function]
+async fn has_vc_arg(arg: Vc<i32>) -> Result<Vc<i32>> {
+    Ok(Vc::cell(*arg.await?))
+}

turbopack/crates/turbo-tasks/src/manager.rs

@@ -306,7 +306,7 @@ pub struct UpdateInfo {
     placeholder_for_future_fields: (),
 }
 
-#[derive(Clone, Copy, Debug, Eq, PartialEq)]
+#[derive(Clone, Copy, Debug, Eq, PartialEq, Hash, Serialize, Deserialize)]
 pub enum TaskPersistence {
     /// Tasks that may be persisted across sessions using serialization.
     Persistent,
@@ -796,7 +796,7 @@ impl<B: Backend + 'static> TurboTasks<B> {
         #[cfg(not(feature = "tokio_tracing"))]
         tokio::task::spawn(future);
 
-        RawVc::LocalOutput(parent_task_id, local_task_id)
+        RawVc::LocalOutput(parent_task_id, persistence, local_task_id)
     }
 
     fn begin_primary_job(&self) {

turbopack/crates/turbo-tasks/src/raw_vc.rs

@@ -12,7 +12,7 @@ use crate::{
     manager::{read_local_output, read_task_cell, read_task_output, with_turbo_tasks},
     registry::{self, get_value_type},
     turbo_tasks, CollectiblesSource, ReadCellOptions, ReadConsistency, ResolvedVc, TaskId,
-    TraitTypeId, ValueType, ValueTypeId, VcValueTrait,
+    TaskPersistence, TraitTypeId, ValueType, ValueTypeId, VcValueTrait,
 };
 
 #[derive(Error, Debug)]
@@ -57,31 +57,34 @@ impl Display for CellId {
 pub enum RawVc {
     TaskOutput(TaskId),
     TaskCell(TaskId, CellId),
-    LocalOutput(TaskId, LocalTaskId),
+    LocalOutput(TaskId, TaskPersistence, LocalTaskId),
 }
 
 impl RawVc {
     pub fn is_resolved(&self) -> bool {
         match self {
-            RawVc::TaskOutput(_) => false,
-            RawVc::TaskCell(_, _) => true,
-            RawVc::LocalOutput(_, _) => false,
+            RawVc::TaskOutput(..) => false,
+            RawVc::TaskCell(..) => true,
+            RawVc::LocalOutput(..) => false,
         }
     }
 
     pub fn is_local(&self) -> bool {
         match self {
-            RawVc::TaskOutput(_) => false,
-            RawVc::TaskCell(_, _) => false,
-            RawVc::LocalOutput(_, _) => true,
+            RawVc::TaskOutput(..) => false,
+            RawVc::TaskCell(..) => false,
+            RawVc::LocalOutput(..) => true,
         }
     }
 
+    /// Returns `true` if the task this `RawVc` reads from cannot be serialized and will not be
+    /// stored in the persistent cache.
+    ///
+    /// See [`TaskPersistence`] for more details.
     pub fn is_transient(&self) -> bool {
         match self {
-            RawVc::TaskOutput(task) | RawVc::TaskCell(task, _) | RawVc::LocalOutput(task, _) => {
-                task.is_transient()
-            }
+            RawVc::TaskOutput(task) | RawVc::TaskCell(task, ..) => task.is_transient(),
+            RawVc::LocalOutput(_, persistence, ..) => *persistence == TaskPersistence::Transient,
         }
     }
 
@@ -145,7 +148,7 @@ impl RawVc {
                         return Err(ResolveTypeError::NoContent);
                     }
                 }
-                RawVc::LocalOutput(task_id, local_task_id) => {
+                RawVc::LocalOutput(task_id, _persistence, local_task_id) => {
                     current = read_local_output(&*tt, task_id, local_task_id)
                         .await
                         .map_err(|source| ResolveTypeError::TaskError { source })?;
@@ -182,7 +185,7 @@ impl RawVc {
                     consistency = ReadConsistency::Eventual;
                 }
                 RawVc::TaskCell(_, _) => return Ok(current),
-                RawVc::LocalOutput(task_id, local_task_id) => {
+                RawVc::LocalOutput(task_id, _persistence, local_task_id) => {
                     debug_assert_eq!(consistency, ReadConsistency::Eventual);
                     current = read_local_output(&*tt, task_id, local_task_id).await?;
                 }
@@ -197,7 +200,7 @@ impl RawVc {
         let mut current = self;
         loop {
             match current {
-                RawVc::LocalOutput(task_id, local_task_id) => {
+                RawVc::LocalOutput(task_id, _persistence, local_task_id) => {
                     current = read_local_output(&*tt, task_id, local_task_id).await?;
                 }
                 non_local => return Ok(non_local),
@@ -212,14 +215,14 @@ impl RawVc {
 
     pub fn get_task_id(&self) -> TaskId {
         match self {
-            RawVc::TaskOutput(t) | RawVc::TaskCell(t, _) | RawVc::LocalOutput(t, _) => *t,
+            RawVc::TaskOutput(t) | RawVc::TaskCell(t, _) | RawVc::LocalOutput(t, ..) => *t,
         }
     }
 
     pub fn try_get_type_id(&self) -> Option<ValueTypeId> {
         match self {
             RawVc::TaskCell(_, CellId { type_id, .. }) => Some(*type_id),
-            RawVc::TaskOutput(_) | RawVc::LocalOutput(_, _) => None,
+            RawVc::TaskOutput(_) | RawVc::LocalOutput(..) => None,
         }
     }
 
@@ -357,7 +360,7 @@ impl Future for ReadRawVcFuture {
                             Err(err) => return Poll::Ready(Err(err)),
                         }
                     }
-                    RawVc::LocalOutput(task_id, local_output_id) => {
+                    RawVc::LocalOutput(task_id, _persistence, local_output_id) => {
                         debug_assert_eq!(this.consistency, ReadConsistency::Eventual);
                         let read_result = tt.try_read_local_output(task_id, local_output_id);
                         match read_result {

turbopack/crates/turbo-tasks/src/task/task_input.rs

@@ -139,7 +139,7 @@ where
     }
 
     fn is_transient(&self) -> bool {
-        self.node.get_task_id().is_transient()
+        self.node.is_transient()
     }
 
     async fn resolve_input(&self) -> Result<Self> {