Repository: vercel/next.js
Pull request: Turbopack: fix incorrect graph state bug

### What?

Technically a task should start with `Dirty` flag set and propagate that info through the aggregation graph. But that would be very expensive to do in the initial compilation. So we have the extra check of tasks not having Output set.

But there is this rare case where a task A calls a task B purely for side effects without `await`ing the output. In this case A would become clean, while B is in its initial computation and don't have Dirty set. This would lead to the subgraph being incorrectly considered as clean.

To fix that case, we eagerly mark all children as Dirty when they don't have Output set, once the parent task gets its Output set.

But there was a bug in that implementation where we early exit when Output is set to something equal the previous Output.

Closes PACK-5024

Changed files:
filename: turbopack/crates/turbo-tasks-backend/src/backend/operation/update_output.rs
patch: @@ -48,86 +48,86 @@ impl UpdateOutputOperation {
         output: Result<RawVc, TurboTasksExecutionError>,
         mut ctx: impl ExecuteContext,
     ) {
-        let mut task = ctx.task(task_id, TaskDataCategory::All);
-        let Some(InProgressState::InProgress(box InProgressStateInner {
-            stale,
-            new_children,
-            ..
-        })) = get!(task, InProgress)
-        else {
-            panic!("Task is not in progress while updating the output");
-        };
-        if *stale {
-            // Skip updating the output when the task is stale
-            return;
-        }
-        let children = if ctx.should_track_children() {
-            new_children.iter().copied().collect()
-        } else {
-            Default::default()
-        };
+        let mut dependent_tasks = Default::default();
+        let mut children = Default::default();
+        let mut queue = AggregationUpdateQueue::new();
 
-        let current_output = get!(task, Output);
-        let output_value = match output {
-            Ok(RawVc::TaskOutput(output_task_id)) => {
-                if let Some(OutputValue::Output(current_task_id)) = current_output
-                    && *current_task_id == output_task_id
-                {
-                    return;
-                }
-                OutputValue::Output(output_task_id)
+        'output: {
+            let mut task = ctx.task(task_id, TaskDataCategory::All);
+            let Some(InProgressState::InProgress(box InProgressStateInner {
+                stale,
+                new_children,
+                ..
+            })) = get!(task, InProgress)
+            else {
+                panic!("Task is not in progress while updating the output");
+            };
+            if *stale {
+                // Skip updating the output when the task is stale
+                break 'output;
             }
-            Ok(RawVc::TaskCell(output_task_id, cell)) => {
-                if let Some(OutputValue::Cell(CellRef {
-                    task: current_task_id,
-                    cell: current_cell,
-                })) = current_output
-                    && *current_task_id == output_task_id
-                    && *current_cell == cell
-                {
-                    return;
-                }
-                OutputValue::Cell(CellRef {
-                    task: output_task_id,
-                    cell,
-                })
+            if ctx.should_track_children() {
+                children = new_children.iter().copied().collect();
             }
-            Ok(RawVc::LocalOutput(..)) => {
-                panic!("Non-local tasks must not return a local Vc");
-            }
-            Err(err) => {
-                if let Some(OutputValue::Error(old_error)) = current_output
-                    && old_error == &err
-                {
-                    return;
-                }
-                OutputValue::Error(err)
-            }
-        };
-        let old_content = task.insert(CachedDataItem::Output {
-            value: output_value,
-        });
 
-        let dependent_tasks = if ctx.should_track_dependencies() {
-            get_many!(task, OutputDependent { task } => task)
-        } else {
-            Default::default()
-        };
+            let current_output = get!(task, Output);
+            let output_value = match output {
+                Ok(RawVc::TaskOutput(output_task_id)) => {
+                    if let Some(OutputValue::Output(current_task_id)) = current_output
+                        && *current_task_id == output_task_id
+                    {
+                        break 'output;
+                    }
+                    OutputValue::Output(output_task_id)
+                }
+                Ok(RawVc::TaskCell(output_task_id, cell)) => {
+                    if let Some(OutputValue::Cell(CellRef {
+                        task: current_task_id,
+                        cell: current_cell,
+                    })) = current_output
+                        && *current_task_id == output_task_id
+                        && *current_cell == cell
+                    {
+                        break 'output;
+                    }
+                    OutputValue::Cell(CellRef {
+                        task: output_task_id,
+                        cell,
+                    })
+                }
+                Ok(RawVc::LocalOutput(..)) => {
+                    panic!("Non-local tasks must not return a local Vc");
+                }
+                Err(err) => {
+                    if let Some(OutputValue::Error(old_error)) = current_output
+                        && old_error == &err
+                    {
+                        break 'output;
+                    }
+                    OutputValue::Error(err)
+                }
+            };
+            let old_content = task.insert(CachedDataItem::Output {
+                value: output_value,
+            });
 
-        let mut queue = AggregationUpdateQueue::new();
+            if ctx.should_track_dependencies() {
+                dependent_tasks = get_many!(task, OutputDependent { task } => task);
+            }
 
-        make_task_dirty_internal(
-            &mut task,
-            task_id,
-            false,
-            #[cfg(feature = "trace_task_dirty")]
-            TaskDirtyCause::InitialDirty,
-            &mut queue,
-            &ctx,
-        );
+            make_task_dirty_internal(
+                &mut task,
+                task_id,
+                false,
+                #[cfg(feature = "trace_task_dirty")]
+                TaskDirtyCause::InitialDirty,
+                &mut queue,
+                &ctx,
+            );
 
-        drop(task);
-        drop(old_content);
+            drop(task);
+            drop(old_content);
+        }
 
         UpdateOutputOperation::MakeDependentTasksDirty {
             #[cfg(feature = "trace_task_dirty")]

filename: turbopack/crates/turbo-tasks-backend/tests/bug2.rs
patch: @@ -0,0 +1,156 @@
+#![feature(arbitrary_self_types)]
+#![feature(arbitrary_self_types_pointers)]
+#![allow(clippy::needless_return)] // tokio macro-generated code doesn't respect this
+
+use std::sync::Arc;
+
+use anyhow::Result;
+use serde::{Deserialize, Serialize};
+use turbo_tasks::{NonLocalValue, State, TaskInput, Vc, trace::TraceRawVcs};
+use turbo_tasks_testing::{Registration, register, run};
+
+static REGISTRATION: Registration = register!();
+
+#[derive(
+    Clone, Debug, PartialEq, Eq, Hash, NonLocalValue, Serialize, Deserialize, TraceRawVcs, TaskInput,
+)]
+pub struct TaskReferenceSpec {
+    task: u16,
+    chain: u8,
+    read: bool,
+    read_strongly_consistent: bool,
+}
+
+#[derive(
+    Clone, Debug, PartialEq, Eq, Hash, NonLocalValue, Serialize, Deserialize, TraceRawVcs, TaskInput,
+)]
+pub struct TaskSpec {
+    references: Vec<TaskReferenceSpec>,
+    children: u8,
+    change: Option<Box<TaskSpec>>,
+}
+
+#[turbo_tasks::value(transparent)]
+struct Iteration(State<usize>);
+
+#[tokio::test]
+async fn graph_bug() {
+    run(&REGISTRATION, move || async move {
+        let spec = vec![
+            TaskSpec {
+                references: vec![TaskReferenceSpec {
+                    task: 1,
+                    chain: 0,
+                    read: false,
+                    read_strongly_consistent: false,
+                }],
+                children: 0,
+                change: Some(Box::new(TaskSpec {
+                    references: vec![TaskReferenceSpec {
+                        task: 1,
+                        chain: 254,
+                        read: false,
+                        read_strongly_consistent: false,
+                    }],
+                    children: 0,
+                    change: None,
+                })),
+            },
+            TaskSpec {
+                references: vec![],
+                children: 0,
+                change: None,
+            },
+        ];
+
+        let it = create_iteration().resolve().await?;
+        it.await?.set(0);
+        println!("🚀 Initial");
+        let task = run_task(Arc::new(spec), it, 0);
+        task.strongly_consistent().await?;
+        println!("🚀 Set iteration to 1");
+        it.await?.set(1);
+        task.strongly_consistent().await?;
+        println!("🚀 Finished strongly consistent wait");
+
+        anyhow::Ok(())
+    })
+    .await
+    .unwrap()
+}
+
+#[turbo_tasks::function]
+fn create_iteration() -> Vc<Iteration> {
+    Vc::cell(State::new(0))
+}
+
+#[turbo_tasks::function]
+async fn run_task_chain(
+    spec: Arc<Vec<TaskSpec>>,
+    iteration: Vc<Iteration>,
+    from: u16,
+    ref_index: usize,
+    to: u16,
+    chain: u8,
+) -> Result<Vc<()>> {
+    println!("run_task_chain(from: {from}, ref_index: {ref_index}, to: {to}, chain: {chain})");
+    if chain > 0 {
+        run_task_chain(spec, iteration, from, ref_index, to, chain - 1).await?;
+    } else {
+        run_task(spec, iteration, to).await?;
+    }
+    Ok(Vc::cell(()))
+}
+
+#[turbo_tasks::function]
+async fn run_task(
+    spec: Arc<Vec<TaskSpec>>,
+    iteration: Vc<Iteration>,
+    task_index: u16,
+) -> Result<Vc<()>> {
+    println!("run_task(task_index: {task_index})");
+    let mut task = &spec[task_index as usize];
+    if task.change.is_some() {
+        let iteration = iteration.await?;
+        let it = *iteration.get();
+        for _ in 0..it {
+            task = if let Some(change) = &task.change {
+                change
+            } else {
+                task
+            };
+        }
+    }
+    for i in 0..task.children {
+        run_task_child(task_index, i).await?;
+    }
+    for (i, reference) in task.references.iter().enumerate() {
+        let call = if reference.chain > 0 {
+            run_task_chain(
+                spec.clone(),
+                iteration,
+                task_index,
+                i,
+                reference.task,
+                reference.chain,
+            )
+        } else {
+            run_task(spec.clone(), iteration, reference.task)
+        };
+        if reference.read {
+            call.await?;
+        }
+        if reference.read_strongly_consistent {
+            call.strongly_consistent().await?;
+        }
+    }
+    Ok(Vc::cell(()))
+}
+
+#[turbo_tasks::function]
+async fn run_task_child(from: u16, i: u8) -> Result<Vc<()>> {
+    println!("run_task_child(from: {from}, i: {i})");
+    let _ = from;
+    let _ = i;
+    Ok(Vc::cell(()))
+}