fix our verify features in turbo-tasks backend

Author: lukesandberg

turbopack/crates/turbo-tasks-backend/Cargo.toml

@@ -13,13 +13,13 @@ bench = false
 workspace = true
 
 [features]
-default = []
+default = ["verify_aggregation_graph", "verify_immutable", "verify_determinism"]
 print_cache_item_size = ["dep:lzzzz"]
 no_fast_stale = []
 verify_serialization = []
 verify_aggregation_graph = []
 verify_immutable = []
-verify_determinism = []
+verify_determinism = ["turbo-tasks/verify_determinism"]
 trace_aggregation_update_queue = []
 trace_aggregation_update = ["trace_aggregation_update_queue"]
 trace_leaf_distance_update = []

turbopack/crates/turbo-tasks-backend/src/backend/mod.rs

@@ -1937,6 +1937,7 @@ impl<B: BackingStorage> TurboTasksBackendInner<B> {
         task_id: TaskId,
         result: Result<RawVc, TurboTasksExecutionError>,
         cell_counters: &AutoMap<ValueTypeId, u32, BuildHasherDefault<FxHasher>, 8>,
+        #[cfg(feature = "verify_determinism")] stateful: bool,
         has_invalidator: bool,
         turbo_tasks: &dyn TurboTasksBackendApi<TurboTasksBackend<B>>,
     ) -> bool {
@@ -1994,6 +1995,8 @@ impl<B: BackingStorage> TurboTasksBackendInner<B> {
             task_id,
             result,
             cell_counters,
+            #[cfg(feature = "verify_determinism")]
+            stateful,
             has_invalidator,
         )
         else {
@@ -2068,6 +2071,7 @@ impl<B: BackingStorage> TurboTasksBackendInner<B> {
         task_id: TaskId,
         result: Result<RawVc, TurboTasksExecutionError>,
         cell_counters: &AutoMap<ValueTypeId, u32, BuildHasherDefault<FxHasher>, 8>,
+        #[cfg(feature = "verify_determinism")] stateful: bool,
         has_invalidator: bool,
     ) -> Option<TaskExecutionCompletePrepareResult> {
         let mut task = ctx.task(task_id, TaskDataCategory::All);
@@ -2132,6 +2136,12 @@ impl<B: BackingStorage> TurboTasksBackendInner<B> {
         // take the children from the task to process them
         let mut new_children = take(new_children);
 
+        // handle stateful (only tracked when verify_determinism is enabled)
+        #[cfg(feature = "verify_determinism")]
+        if stateful {
+            task.set_stateful(true);
+        }
+
         // handle has_invalidator
         if has_invalidator {
             task.set_invalidator(true);
@@ -2575,7 +2585,8 @@ impl<B: BackingStorage> TurboTasksBackendInner<B> {
         };
 
         // Update the dirty state
-        if old_dirtyness != new_dirtyness {
+        let dirty_changed = old_dirtyness != new_dirtyness;
+        if dirty_changed {
             if let Some(value) = new_dirtyness {
                 task.set_dirty(value);
             } else if old_dirtyness.is_some() {
@@ -3139,7 +3150,7 @@ impl<B: BackingStorage> TurboTasksBackendInner<B> {
                         "Task {} {} doesn't report to any root but is reachable from one (uppers: \
                          {:?})",
                         task_id,
-                        ctx.get_task_description(task_id),
+                        task.get_task_description(),
                         uppers
                     );
                 }
@@ -3150,14 +3161,14 @@ impl<B: BackingStorage> TurboTasksBackendInner<B> {
                     .collect();
                 for collectible in aggregated_collectibles {
                     collectibles
-                        .entry(collectible)
+                        .entry(*collectible)
                         .or_insert_with(|| (false, Vec::new()))
                         .1
                         .push(task_id);
                 }
 
                 let own_collectibles: Vec<_> = task
-                    .iter_collectibles_entries()
+                    .iter_collectibles()
                     .filter_map(|(&collectible, &value)| (value > 0).then_some(collectible))
                     .collect::<Vec<_>>();
                 for collectible in own_collectibles {
@@ -3195,22 +3206,25 @@ impl<B: BackingStorage> TurboTasksBackendInner<B> {
 
                 if should_be_in_upper {
                     for upper_id in uppers {
-                        let task = ctx.task(upper_id, TaskDataCategory::All);
-                        let in_upper = task
+                        let upper = ctx.task(upper_id, TaskDataCategory::All);
+                        let in_upper = upper
                             .get_aggregated_dirty_containers(&task_id)
                             .is_some_and(|&dirty| dirty > 0);
                         if !in_upper {
-                            let containers: Vec<_> = task
-                                .iter_aggregated_dirty_containers_entries()
+                            let containers: Vec<_> = upper
+                                .iter_aggregated_dirty_containers()
                                 .map(|(&k, &v)| (k, v))
                                 .collect();
+                            let upper_task_desc = upper.get_task_description();
+                            drop(upper);
                             panic!(
                                 "Task {} ({}) is dirty, but is not listed in the upper task {} \
                                  ({})\nThese dirty containers are present:\n{:#?}",
                                 task_id,
-                                ctx.get_task_description(task_id),
+                                ctx.task(task_id, TaskDataCategory::Data)
+                                    .get_task_description(),
                                 upper_id,
-                                ctx.get_task_description(upper_id),
+                                upper_task_desc,
                                 containers,
                             );
                         }
@@ -3229,7 +3243,10 @@ impl<B: BackingStorage> TurboTasksBackendInner<B> {
                         collectible,
                         task_ids
                             .iter()
-                            .map(|t| format!("{t} {}", ctx.get_task_description(*t)))
+                            .map(|t| format!(
+                                "{t} {}",
+                                ctx.task(*t, TaskDataCategory::Data).get_task_description()
+                            ))
                             .collect::<Vec<_>>()
                     )
                     .unwrap();
@@ -3245,8 +3262,8 @@ impl<B: BackingStorage> TurboTasksBackendInner<B> {
                         writeln!(stdout, "{task_id:?} -> {upper_id:?}").unwrap();
                     }
                     while let Some(task_id) = queue.pop() {
-                        let desc = ctx.get_task_description(task_id);
                         let task = ctx.task(task_id, TaskDataCategory::All);
+                        let desc = task.get_task_description();
                         let aggregated_collectible = task
                             .get_aggregated_collectibles(&collectible)
                             .copied()
@@ -3436,13 +3453,16 @@ impl<B: BackingStorage> Backend for TurboTasksBackend<B> {
         task_id: TaskId,
         result: Result<RawVc, TurboTasksExecutionError>,
         cell_counters: &AutoMap<ValueTypeId, u32, BuildHasherDefault<FxHasher>, 8>,
+        #[cfg(feature = "verify_determinism")] stateful: bool,
         has_invalidator: bool,
         turbo_tasks: &dyn TurboTasksBackendApi<Self>,
     ) -> bool {
         self.0.task_execution_completed(
             task_id,
             result,
             cell_counters,
+            #[cfg(feature = "verify_determinism")]
+            stateful,
             has_invalidator,
             turbo_tasks,
         )

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

@@ -55,8 +55,10 @@ impl UpdateCellOperation {
         content: CellContent,
         is_serializable_cell_content: bool,
         updated_key_hashes: Option<SmallVec<[u64; 2]>>,
-        #[cfg(feature = "verify_determinism")] verification_mode: VerificationMode,
-        #[cfg(not(feature = "verify_determinism"))] _verification_mode: VerificationMode,
+        #[cfg(feature = "verify_determinism")]
+        verification_mode: turbo_tasks::backend::VerificationMode,
+        #[cfg(not(feature = "verify_determinism"))]
+        _verification_mode: turbo_tasks::backend::VerificationMode,
         mut ctx: impl ExecuteContext<'_>,
     ) {
         let content = if let CellContent(Some(new_content)) = content {
@@ -80,11 +82,14 @@ impl UpdateCellOperation {
 
                 // Check if this assumption holds.
                 #[cfg(feature = "verify_determinism")]
-                if !is_stateful
-                    && matches!(verification_mode, VerificationMode::EqualityCheck)
+                if !task.stateful()
+                    && matches!(
+                        verification_mode,
+                        turbo_tasks::backend::VerificationMode::EqualityCheck
+                    )
                     && content != task.get_cell_data(is_serializable_cell_content, cell)
                 {
-                    let task_description = ctx.get_task_description(task_id);
+                    let task_description = task.get_task_description();
                     let cell_type = turbo_tasks::registry::get_value_type(cell.type_id).global_name;
                     eprintln!(
                         "Task {} updated cell #{} (type: {}) while recomputing",

turbopack/crates/turbo-tasks-backend/src/backend/storage_schema.rs

@@ -181,6 +181,12 @@ struct TaskStorageSchema {
     #[field(storage = "flag", category = "transient")]
     prefetched: bool,
 
+    /// Whether this task has allocated a State (has interior mutability).
+    /// Only set when `verify_determinism`` feature is enabled.
+    /// Used to skip determinism checks for stateful tasks.
+    #[field(storage = "flag", category = "transient")]
+    stateful: bool,
+
     // =========================================================================
     // CHILDREN & AGGREGATION (meta)
     // =========================================================================

turbopack/crates/turbo-tasks/Cargo.toml

@@ -13,6 +13,7 @@ default = []
 tokio_tracing = ["tokio/tracing"]
 hanging_detection = []
 task_id_details = []
+verify_determinism = []
 
 # TODO(bgw): This feature is here to unblock turning on local tasks by default. It's only turned on
 # in unit tests. This will be removed very soon.

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

@@ -493,6 +493,7 @@ pub trait Backend: Sync + Send {
         task: TaskId,
         result: Result<RawVc, TurboTasksExecutionError>,
         cell_counters: &AutoMap<ValueTypeId, u32, BuildHasherDefault<FxHasher>, 8>,
+        #[cfg(feature = "verify_determinism")] stateful: bool,
         has_invalidator: bool,
         turbo_tasks: &dyn TurboTasksBackendApi<Self>,
     ) -> bool;

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

@@ -116,8 +116,8 @@ pub use crate::{
         CurrentCellRef, ReadCellTracking, ReadConsistency, ReadTracking, TaskPersistence,
         TaskPriority, TurboTasks, TurboTasksApi, TurboTasksBackendApi, TurboTasksCallApi, Unused,
         UpdateInfo, dynamic_call, emit, get_serialization_invalidator, mark_finished, mark_root,
-        mark_session_dependent, prevent_gc, run, run_once, run_once_with_reason, trait_call,
-        turbo_tasks, turbo_tasks_scope, turbo_tasks_weak, with_turbo_tasks,
+        mark_session_dependent, mark_stateful, prevent_gc, run, run_once, run_once_with_reason,
+        trait_call, turbo_tasks, turbo_tasks_scope, turbo_tasks_weak, with_turbo_tasks,
     },
     mapped_read_ref::MappedReadRef,
     output::OutputContent,

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

@@ -518,6 +518,11 @@ struct CurrentTaskState {
     execution_id: ExecutionId,
     priority: TaskPriority,
 
+    /// True if the current task has state in cells (interior mutability).
+    /// Only tracked when verify_determinism feature is enabled.
+    #[cfg(feature = "verify_determinism")]
+    stateful: bool,
+
     /// True if the current task uses an external invalidator
     has_invalidator: bool,
 
@@ -541,6 +546,8 @@ impl CurrentTaskState {
             task_id: Some(task_id),
             execution_id,
             priority,
+            #[cfg(feature = "verify_determinism")]
+            stateful: false,
             has_invalidator: false,
             cell_counters: Some(AutoMap::default()),
             local_tasks: Vec::new(),
@@ -553,6 +560,8 @@ impl CurrentTaskState {
             task_id: None,
             execution_id,
             priority,
+            #[cfg(feature = "verify_determinism")]
+            stateful: false,
             has_invalidator: false,
             cell_counters: None,
             local_tasks: Vec::new(),
@@ -1142,14 +1151,28 @@ impl<B: Backend + 'static> TurboTasks<B> {
     }
 
     fn finish_current_task_state(&self) -> FinishedTaskState {
+        #[cfg(feature = "verify_determinism")]
+        let (stateful, has_invalidator) = CURRENT_TASK_STATE.with(|cell| {
+            let CurrentTaskState {
+                stateful,
+                has_invalidator,
+                ..
+            } = &mut *cell.write().unwrap();
+            (*stateful, *has_invalidator)
+        });
+        #[cfg(not(feature = "verify_determinism"))]
         let has_invalidator = CURRENT_TASK_STATE.with(|cell| {
             let CurrentTaskState {
                 has_invalidator, ..
             } = &mut *cell.write().unwrap();
             *has_invalidator
         });
 
-        FinishedTaskState { has_invalidator }
+        FinishedTaskState {
+            #[cfg(feature = "verify_determinism")]
+            stateful,
+            has_invalidator,
+        }
     }
 
     pub fn backend(&self) -> &B {
@@ -1208,6 +1231,12 @@ impl<B: Backend> Executor<TurboTasks<B>, ScheduledTask, TaskPriority> for TurboT
                                     Err(err) => Err(TurboTasksExecutionError::Panic(Arc::new(err))),
                                 };
 
+                                #[cfg(feature = "verify_determinism")]
+                                let FinishedTaskState {
+                                    stateful,
+                                    has_invalidator,
+                                } = this.finish_current_task_state();
+                                #[cfg(not(feature = "verify_determinism"))]
                                 let FinishedTaskState { has_invalidator } =
                                     this.finish_current_task_state();
                                 let cell_counters = CURRENT_TASK_STATE
@@ -1216,6 +1245,8 @@ impl<B: Backend> Executor<TurboTasks<B>, ScheduledTask, TaskPriority> for TurboT
                                     task_id,
                                     result,
                                     &cell_counters,
+                                    #[cfg(feature = "verify_determinism")]
+                                    stateful,
                                     has_invalidator,
                                     &*this,
                                 )
@@ -1310,6 +1341,11 @@ impl<B: Backend> Executor<TurboTasks<B>, ScheduledTask, TaskPriority> for TurboT
 }
 
 struct FinishedTaskState {
+    /// True if the task has state in cells (interior mutability).
+    /// Only tracked when verify_determinism feature is enabled.
+    #[cfg(feature = "verify_determinism")]
+    stateful: bool,
+
     /// True if the task uses an external invalidator
     has_invalidator: bool,
 }
@@ -1849,10 +1885,22 @@ pub fn mark_finished() {
 }
 
 /// Returns a [`SerializationInvalidator`] that can be used to invalidate the
-/// serialization of the current task cells
+/// serialization of the current task cells.
+///
+/// Also marks the current task as stateful when the `verify_determinism` feature is enabled,
+/// since State allocation implies interior mutability.
 pub fn get_serialization_invalidator() -> SerializationInvalidator {
     CURRENT_TASK_STATE.with(|cell| {
-        let CurrentTaskState { task_id, .. } = &mut *cell.write().unwrap();
+        let CurrentTaskState {
+            task_id,
+            #[cfg(feature = "verify_determinism")]
+            stateful,
+            ..
+        } = &mut *cell.write().unwrap();
+        #[cfg(feature = "verify_determinism")]
+        {
+            *stateful = true;
+        }
         let Some(task_id) = *task_id else {
             panic!(
                 "get_serialization_invalidator() can only be used in the context of a turbo_tasks \
@@ -1872,6 +1920,24 @@ pub fn mark_invalidator() {
     })
 }
 
+/// Marks the current task as stateful. This is used to indicate that the task
+/// has interior mutability (e.g., via State or TransientState), which means
+/// the task may produce different outputs even with the same inputs.
+///
+/// Only has an effect when the `verify_determinism` feature is enabled.
+#[cfg(feature = "verify_determinism")]
+pub fn mark_stateful() {
+    CURRENT_TASK_STATE.with(|cell| {
+        let CurrentTaskState { stateful, .. } = &mut *cell.write().unwrap();
+        *stateful = true;
+    })
+}
+
+#[cfg(not(feature = "verify_determinism"))]
+pub fn mark_stateful() {
+    // No-op when verify_determinism is not enabled
+}
+
 pub fn prevent_gc() {
     // TODO implement garbage collection
 }

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

@@ -13,7 +13,7 @@ use tracing::trace_span;
 use crate::{
     Invalidator, OperationValue, SerializationInvalidator, get_invalidator,
     get_serialization_invalidator, manager::with_turbo_tasks, mark_session_dependent,
-    trace::TraceRawVcs,
+    mark_stateful, trace::TraceRawVcs,
 };
 
 #[derive(Encode, Decode)]
@@ -333,6 +333,7 @@ impl<T> Eq for TransientState<T> {}
 
 impl<T> TransientState<T> {
     pub fn new() -> Self {
+        mark_stateful();
         Self {
             inner: Mutex::new(StateInner::new(None)),
         }