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

Using the parent task id for determining the persistence of a `RawVc::LocalOutput` is (sometimes) wrong.

If we have a transient task calling a persistent task, the wrapped `LocalOutput` should be persistent.

We just store the persistence here, and not the task id of the wrapped `LocalOutput`, because when this is constructed, no non-local task id has been reserved yet. We don't really need the full task id though: just the persistence/transient bit is sufficient.

After this PR stack, the parent task id isn't really used for anything anymore, so we can get rid of it (TODO).

Tested using the repro here: https://vercel.slack.com/archives/C046HAU4H7F/p1744792143917369?thread_ts=1744792143.917369

Author: bgw

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

@@ -11,8 +11,8 @@ use crate::{
     id::LocalTaskId,
     manager::{read_local_output, read_task_cell, read_task_output, with_turbo_tasks},
     registry::{self, get_value_type},
-    turbo_tasks, CollectiblesSource, ReadCellOptions, ReadConsistency, TaskId, TraitTypeId,
-    ValueType, ValueTypeId, Vc, VcValueTrait,
+    turbo_tasks, CollectiblesSource, ReadCellOptions, ReadConsistency, TaskId, TaskPersistence,
+    TraitTypeId, ValueType, ValueTypeId, Vc, 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(crate) 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(crate) 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,7 +215,7 @@ 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,
         }
     }
 
@@ -347,7 +350,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

@@ -141,7 +141,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> {