Turbopack: add verify_determinism feature to check if tasks are deterministic (#85559)

### What?

Enabling the feature will double execute tasks and print to stdout when tasks are updating cells while recomputing

The change will also change the behavior of cell updates during recomputation slightly. Instead of updating the cell and not notifying dependents, it will not update the cell at all ignoring the (unexpected) change.
This means when a task is recomputed without being invalidated, it won't update any cell data, except for the missing cell values.

Author: sokra

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

@@ -19,6 +19,7 @@ no_fast_stale = []
 verify_serialization = []
 verify_aggregation_graph = []
 verify_immutable = []
+verify_determinism = []
 trace_aggregation_update = []
 trace_find_and_schedule = []
 trace_task_completion = []

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

@@ -31,7 +31,7 @@ use turbo_tasks::{
     TraitTypeId, TurboTasksBackendApi, ValueTypeId,
     backend::{
         Backend, CachedTaskType, CellContent, TaskExecutionSpec, TransientTaskRoot,
-        TransientTaskType, TurboTasksExecutionError, TypedCellContent,
+        TransientTaskType, TurboTasksExecutionError, TypedCellContent, VerificationMode,
     },
     event::{Event, EventListener},
     message_queue::TimingEvent,
@@ -423,6 +423,8 @@ struct TaskExecutionCompletePrepareResult {
     pub new_children: FxHashSet<TaskId>,
     pub removed_data: Vec<CachedDataItem>,
     pub is_now_immutable: bool,
+    #[cfg(feature = "verify_determinism")]
+    pub no_output_set: bool,
     pub new_output: Option<OutputValue>,
     pub output_dependent_tasks: SmallVec<[TaskId; 4]>,
 }
@@ -1765,6 +1767,8 @@ impl<B: BackingStorage> TurboTasksBackendInner<B> {
             new_children,
             mut removed_data,
             is_now_immutable,
+            #[cfg(feature = "verify_determinism")]
+            no_output_set,
             new_output,
             output_dependent_tasks,
         }) = self.task_execution_completed_prepare(
@@ -1809,6 +1813,8 @@ impl<B: BackingStorage> TurboTasksBackendInner<B> {
         if self.task_execution_completed_finish(
             &mut ctx,
             task_id,
+            #[cfg(feature = "verify_determinism")]
+            no_output_set,
             new_output,
             &mut removed_data,
             is_now_immutable,
@@ -1843,6 +1849,8 @@ impl<B: BackingStorage> TurboTasksBackendInner<B> {
                 new_children: Default::default(),
                 removed_data: Default::default(),
                 is_now_immutable: false,
+                #[cfg(feature = "verify_determinism")]
+                no_output_set: false,
                 new_output: None,
                 output_dependent_tasks: Default::default(),
             });
@@ -2021,6 +2029,8 @@ impl<B: BackingStorage> TurboTasksBackendInner<B> {
 
         // Check if output need to be updated
         let current_output = get!(task, Output);
+        #[cfg(feature = "verify_determinism")]
+        let no_output_set = current_output.is_none();
         let new_output = match result {
             Ok(RawVc::TaskOutput(output_task_id)) => {
                 if let Some(OutputValue::Output(current_task_id)) = current_output
@@ -2092,6 +2102,8 @@ impl<B: BackingStorage> TurboTasksBackendInner<B> {
             new_children,
             removed_data,
             is_now_immutable,
+            #[cfg(feature = "verify_determinism")]
+            no_output_set,
             new_output,
             output_dependent_tasks,
         })
@@ -2232,6 +2244,7 @@ impl<B: BackingStorage> TurboTasksBackendInner<B> {
         &self,
         ctx: &mut impl ExecuteContext<'_>,
         task_id: TaskId,
+        #[cfg(feature = "verify_determinism")] no_output_set: bool,
         new_output: Option<OutputValue>,
         removed_data: &mut Vec<CachedDataItem>,
         is_now_immutable: bool,
@@ -2306,7 +2319,8 @@ impl<B: BackingStorage> TurboTasksBackendInner<B> {
             None
         };
 
-        let data_update = if old_dirty_state != new_dirty_state {
+        let dirty_changed = old_dirty_state != new_dirty_state;
+        let data_update = if dirty_changed {
             if let Some(new_dirty_state) = new_dirty_state {
                 task.insert(CachedDataItem::Dirty {
                     value: new_dirty_state,
@@ -2353,17 +2367,32 @@ impl<B: BackingStorage> TurboTasksBackendInner<B> {
             None
         };
 
-        drop(task);
-        drop(old_content);
+        #[cfg(feature = "verify_determinism")]
+        let reschedule = (dirty_changed || no_output_set) && !task_id.is_transient();
+        #[cfg(not(feature = "verify_determinism"))]
+        let reschedule = false;
+        if reschedule {
+            task.add_new(CachedDataItem::InProgress {
+                value: InProgressState::Scheduled {
+                    done_event,
+                    reason: TaskExecutionReason::Stale,
+                },
+            });
+            drop(task);
+        } else {
+            drop(task);
+
+            // Notify dependent tasks that are waiting for this task to finish
+            done_event.notify(usize::MAX);
+        }
 
-        // Notify dependent tasks that are waiting for this task to finish
-        done_event.notify(usize::MAX);
+        drop(old_content);
 
         if let Some(data_update) = data_update {
             AggregationUpdateQueue::run(data_update, ctx);
         }
 
-        false
+        reschedule
     }
 
     fn task_execution_completed_cleanup(&self, ctx: &mut impl ExecuteContext<'_>, task_id: TaskId) {
@@ -2652,12 +2681,14 @@ impl<B: BackingStorage> TurboTasksBackendInner<B> {
         task_id: TaskId,
         cell: CellId,
         content: CellContent,
+        verification_mode: VerificationMode,
         turbo_tasks: &dyn TurboTasksBackendApi<TurboTasksBackend<B>>,
     ) {
         operation::UpdateCellOperation::run(
             task_id,
             cell,
             content,
+            verification_mode,
             self.execute_context(turbo_tasks),
         );
     }
@@ -3251,9 +3282,11 @@ impl<B: BackingStorage> Backend for TurboTasksBackend<B> {
         task_id: TaskId,
         cell: CellId,
         content: CellContent,
+        verification_mode: VerificationMode,
         turbo_tasks: &dyn TurboTasksBackendApi<Self>,
     ) {
-        self.0.update_task_cell(task_id, cell, content, turbo_tasks);
+        self.0
+            .update_task_cell(task_id, cell, content, verification_mode, turbo_tasks);
     }
 
     fn mark_own_task_as_finished(

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

@@ -2,6 +2,8 @@ use std::mem::take;
 
 use serde::{Deserialize, Serialize};
 use smallvec::SmallVec;
+#[cfg(not(feature = "verify_determinism"))]
+use turbo_tasks::backend::VerificationMode;
 use turbo_tasks::{CellId, TaskId, TypedSharedReference, backend::CellContent};
 
 #[cfg(feature = "trace_task_dirty")]
@@ -13,7 +15,7 @@ use crate::{
             AggregationUpdateQueue, ExecuteContext, Operation, TaskGuard,
             invalidate::make_task_dirty_internal,
         },
-        storage::{get_many, remove},
+        storage::{get, get_many, remove},
     },
     data::{CachedDataItem, CachedDataItemKey, CellRef},
 };
@@ -40,18 +42,61 @@ pub enum UpdateCellOperation {
 }
 
 impl UpdateCellOperation {
-    pub fn run(task_id: TaskId, cell: CellId, content: CellContent, mut ctx: impl ExecuteContext) {
+    pub fn run(
+        task_id: TaskId,
+        cell: CellId,
+        content: CellContent,
+        #[cfg(feature = "verify_determinism")] verification_mode: VerificationMode,
+        #[cfg(not(feature = "verify_determinism"))] _verification_mode: VerificationMode,
+        mut ctx: impl ExecuteContext,
+    ) {
+        let content = if let CellContent(Some(new_content)) = content {
+            Some(new_content.into_typed(cell.type_id))
+        } else {
+            None
+        };
+
         let mut task = ctx.task(task_id, TaskDataCategory::All);
 
+        let is_stateful = task.has_key(&CachedDataItemKey::Stateful {});
         // We need to detect recomputation, because here the content has not actually changed (even
         // if it's not equal to the old content, as not all values implement Eq). We have to
         // assume that tasks are deterministic and pure.
-        let should_invalidate = ctx.should_track_dependencies()
-            && (task.has_key(&CachedDataItemKey::Dirty {}) ||
+        let assume_unchanged = !ctx.should_track_dependencies()
+            || (!task.has_key(&CachedDataItemKey::Dirty {})
             // This is a hack for the streaming hack. Stateful tasks are never recomputed, so this forces invalidation for them in case of this hack.
-            task.has_key(&CachedDataItemKey::Stateful {}));
+            && !is_stateful);
+
+        let old_content = get!(task, CellData { cell });
+
+        if assume_unchanged {
+            if old_content.is_some() {
+                // Never update cells when recomputing if they already have a value.
+                // It's not expected that content changes during recomputation.
+
+                // Check if this assumption holds.
+                #[cfg(feature = "verify_determinism")]
+                if !is_stateful
+                    && matches!(verification_mode, VerificationMode::EqualityCheck)
+                    && content.as_ref() != old_content
+                {
+                    let task_description = ctx.get_task_description(task_id);
+                    let cell_type = turbo_tasks::registry::get_value_type(cell.type_id).global_name;
+                    eprintln!(
+                        "Task {} updated cell #{} (type: {}) while recomputing",
+                        task_description, cell.index, cell_type
+                    );
+                }
+                return;
+            } else {
+                // Initial computation, or computation after a cell has been cleared.
+                // We can just set the content, but we don't want to notify dependent tasks,
+                // as we assume that content hasn't changed (deterministic tasks).
+            }
+        } else {
+            // When not recomputing, we need to notify dependent tasks if the content actually
+            // changes.
 
-        if should_invalidate {
             let dependent_tasks: SmallVec<[TaskId; 4]> = get_many!(
                 task,
                 CellDependent { cell: dependent_cell, task }
@@ -78,12 +123,6 @@ impl UpdateCellOperation {
                 drop(task);
                 drop(old_content);
 
-                let content = if let CellContent(Some(new_content)) = content {
-                    Some(new_content.into_typed(cell.type_id))
-                } else {
-                    None
-                };
-
                 UpdateCellOperation::InvalidateWhenCellDependency {
                     cell_ref: CellRef {
                         task: task_id,
@@ -101,8 +140,7 @@ impl UpdateCellOperation {
         // Fast path: We don't need to invalidate anything.
         // So we can just update the cell content.
 
-        let old_content = if let CellContent(Some(new_content)) = content {
-            let new_content = new_content.into_typed(cell.type_id);
+        let old_content = if let Some(new_content) = content {
             task.insert(CachedDataItem::CellData {
                 cell,
                 value: new_content,

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

@@ -18,6 +18,7 @@ static REGISTRATION: Registration = register!();
 #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
 async fn test_spawns_detached() -> anyhow::Result<()> {
     run_once(&REGISTRATION, || async {
+        println!("test_spawns_detached");
         // HACK: The watch channel we use has an incorrect implementation of `TraceRawVcs`, just
         // disable GC for the test so this can't cause any problems.
         prevent_gc();
@@ -73,7 +74,9 @@ async fn spawns_detached(
     sender: TransientInstance<WatchSenderTaskInput<Option<Vc<u32>>>>,
 ) -> Vc<()> {
     tokio::spawn(turbo_tasks().detached_for_testing(Box::pin(async move {
+        println!("spawns_detached: waiting for notify");
         notify.0.notified().await;
+        println!("spawns_detached: notified, sending value");
         // creating cells after the normal lifetime of the task should be okay, as the parent task
         // is waiting on us before exiting!
         sender.0.send(Some(Vc::cell(42))).unwrap();

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

@@ -18,7 +18,7 @@ use tokio::sync::mpsc::Receiver;
 use turbo_tasks::{
     CellId, ExecutionId, InvalidationReason, LocalTaskId, MagicAny, RawVc, ReadCellOptions,
     ReadOutputOptions, TaskId, TaskPersistence, TraitTypeId, TurboTasksApi, TurboTasksCallApi,
-    backend::{CellContent, TaskCollectiblesMap, TypedCellContent},
+    backend::{CellContent, TaskCollectiblesMap, TypedCellContent, VerificationMode},
     event::{Event, EventListener},
     message_queue::CompilationEvent,
     test_helpers::with_turbo_tasks_for_testing,
@@ -261,7 +261,13 @@ impl TurboTasksApi for VcStorage {
         .into_typed(index.type_id))
     }
 
-    fn update_own_task_cell(&self, task: TaskId, index: CellId, content: CellContent) {
+    fn update_own_task_cell(
+        &self,
+        task: TaskId,
+        index: CellId,
+        content: CellContent,
+        _verification_mode: VerificationMode,
+    ) {
         let mut map = self.cells.lock().unwrap();
         let cell = map.entry((task, index)).or_default();
         *cell = content;

turbopack/crates/turbo-tasks-testing/src/run.rs

@@ -1,4 +1,4 @@
-use std::{fmt::Debug, future::Future, sync::Arc};
+use std::{env, fmt::Debug, future::Future, sync::Arc};
 
 use anyhow::Result;
 use turbo_tasks::{TurboTasksApi, trace::TraceRawVcs};
@@ -98,22 +98,28 @@ where
     F: Future<Output = Result<T>> + Send + 'static,
     T: Debug + PartialEq + Eq + TraceRawVcs + Send + 'static,
 {
+    let single_run = env::var("SINGLE_RUN").is_ok();
     let name = closure_to_name(&fut);
     let tt = registration.create_turbo_tasks(&name, true);
     println!("Run #1 (without cache)");
     let start = std::time::Instant::now();
     let first = fut(tt.clone()).await?;
     println!("Run #1 took {:?}", start.elapsed());
-    for i in 2..10 {
-        println!("Run #{i} (with memory cache, same TurboTasks instance)");
-        let start = std::time::Instant::now();
-        let second = fut(tt.clone()).await?;
-        println!("Run #{i} took {:?}", start.elapsed());
-        assert_eq!(first, second);
+    if !single_run {
+        for i in 2..10 {
+            println!("Run #{i} (with memory cache, same TurboTasks instance)");
+            let start = std::time::Instant::now();
+            let second = fut(tt.clone()).await?;
+            println!("Run #{i} took {:?}", start.elapsed());
+            assert_eq!(first, second);
+        }
     }
     let start = std::time::Instant::now();
     tt.stop_and_wait().await;
     println!("Stopping TurboTasks took {:?}", start.elapsed());
+    if single_run {
+        return Ok(());
+    }
     for i in 10..20 {
         let tt = registration.create_turbo_tasks(&name, false);
         println!("Run #{i} (with filesystem cache if available, new TurboTasks instance)");

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

@@ -497,6 +497,11 @@ impl From<anyhow::Error> for TurboTasksExecutionError {
     }
 }
 
+pub enum VerificationMode {
+    EqualityCheck,
+    Skip,
+}
+
 pub trait Backend: Sync + Send {
     #[allow(unused_variables)]
     fn startup(&self, turbo_tasks: &dyn TurboTasksBackendApi<Self>) {}
@@ -621,6 +626,7 @@ pub trait Backend: Sync + Send {
         task: TaskId,
         index: CellId,
         content: CellContent,
+        verification_mode: VerificationMode,
         turbo_tasks: &dyn TurboTasksBackendApi<Self>,
     );