Turbopack: add task_id_details feature (#84970)

### What?

Add a bunch of feature flags that help debugging cases where it is loosing invalidations.

Author: sokra

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

@@ -24,6 +24,8 @@ trace_aggregation_update = []
 trace_find_and_schedule = []
 trace_task_completion = []
 trace_task_dirty = []
+trace_task_output_dependencies = []
+trace_task_details = []
 lmdb = ["dep:lmdb-rkv"]
 
 [dependencies]

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

@@ -24,7 +24,7 @@ use parking_lot::{Condvar, Mutex};
 use rustc_hash::{FxHashMap, FxHashSet, FxHasher};
 use smallvec::{SmallVec, smallvec};
 use tokio::time::{Duration, Instant};
-use tracing::{Span, field::Empty, info_span, trace_span};
+use tracing::{Span, info_span, trace_span};
 use turbo_tasks::{
     CellId, FxDashMap, FxIndexMap, KeyValuePair, RawVc, ReadCellOptions, ReadConsistency,
     ReadOutputOptions, ReadTracking, SessionId, TRANSIENT_TASK_BIT, TaskExecutionReason, TaskId,
@@ -722,21 +722,23 @@ impl<B: BackingStorage> TurboTasksBackendInner<B> {
             let result = match output {
                 OutputValue::Cell(cell) => Ok(Ok(RawVc::TaskCell(cell.task, cell.cell))),
                 OutputValue::Output(task) => Ok(Ok(RawVc::TaskOutput(*task))),
-                OutputValue::Error(error) => {
-                    let err: anyhow::Error = error.clone().into();
-                    Err(err.context(format!(
-                        "Execution of {} failed",
-                        ctx.get_task_description(task_id)
-                    )))
-                }
+                OutputValue::Error(error) => Err(error
+                    .with_task_context(ctx.get_task_description(task_id), Some(task_id))
+                    .into()),
             };
             if let Some(mut reader_task) = reader_task
                 && options.tracking.should_track(result.is_err())
                 && (!task.is_immutable() || cfg!(feature = "verify_immutable"))
             {
-                let reader = reader.unwrap();
+                #[cfg(feature = "trace_task_output_dependencies")]
+                let _span = tracing::trace_span!(
+                    "add output dependency",
+                    task = %task_id,
+                    dependent_task = ?reader
+                )
+                .entered();
                 let _ = task.add(CachedDataItem::OutputDependent {
-                    task: reader,
+                    task: reader.unwrap(),
                     value: (),
                 });
                 drop(task);
@@ -1763,7 +1765,22 @@ impl<B: BackingStorage> TurboTasksBackendInner<B> {
         // The task might be invalidated during this process, so we need to check the stale flag
         // at the start of every step.
 
-        let span = tracing::trace_span!("task execution completed", immutable = Empty).entered();
+        #[cfg(not(feature = "trace_task_details"))]
+        let _span = tracing::trace_span!("task execution completed").entered();
+        #[cfg(feature = "trace_task_details")]
+        let span = tracing::trace_span!(
+            "task execution completed",
+            task_id = display(task_id),
+            result = match result.as_ref() {
+                Ok(value) => display(either::Either::Left(value)),
+                Err(err) => display(either::Either::Right(err)),
+            },
+            immutable = tracing::field::Empty,
+            new_output = tracing::field::Empty,
+            output_dependents = tracing::field::Empty,
+            stale = tracing::field::Empty,
+        )
+        .entered();
         let mut ctx = self.execute_context(turbo_tasks);
 
         let Some(TaskExecutionCompletePrepareResult {
@@ -1776,6 +1793,7 @@ impl<B: BackingStorage> TurboTasksBackendInner<B> {
             output_dependent_tasks,
         }) = self.task_execution_completed_prepare(
             &mut ctx,
+            #[cfg(feature = "trace_task_details")]
             &span,
             task_id,
             result,
@@ -1785,9 +1803,16 @@ impl<B: BackingStorage> TurboTasksBackendInner<B> {
         )
         else {
             // Task was stale and has been rescheduled
+            #[cfg(feature = "trace_task_details")]
+            span.record("stale", "true");
             return true;
         };
 
+        #[cfg(feature = "trace_task_details")]
+        span.record("new_output", new_output.is_some());
+        #[cfg(feature = "trace_task_details")]
+        span.record("output_dependents", output_dependent_tasks.len());
+
         // When restoring from filesystem cache the following might not be executed (since we can
         // suspend in `CleanupOldEdgesOperation`), but that's ok as the task is still dirty and
         // would be executed again.
@@ -1810,6 +1835,8 @@ impl<B: BackingStorage> TurboTasksBackendInner<B> {
             && self.task_execution_completed_connect(&mut ctx, task_id, new_children)
         {
             // Task was stale and has been rescheduled
+            #[cfg(feature = "trace_task_details")]
+            span.record("stale", "true");
             return true;
         }
 
@@ -1823,6 +1850,8 @@ impl<B: BackingStorage> TurboTasksBackendInner<B> {
             is_now_immutable,
         ) {
             // Task was stale and has been rescheduled
+            #[cfg(feature = "trace_task_details")]
+            span.record("stale", "true");
             return true;
         }
 
@@ -1836,7 +1865,7 @@ impl<B: BackingStorage> TurboTasksBackendInner<B> {
     fn task_execution_completed_prepare(
         &self,
         ctx: &mut impl ExecuteContext<'_>,
-        span: &Span,
+        #[cfg(feature = "trace_task_details")] span: &Span,
         task_id: TaskId,
         result: Result<RawVc, TurboTasksExecutionError>,
         cell_counters: &AutoMap<ValueTypeId, u32, BuildHasherDefault<FxHasher>, 8>,
@@ -2088,6 +2117,7 @@ impl<B: BackingStorage> TurboTasksBackendInner<B> {
         {
             is_now_immutable = true;
         }
+        #[cfg(feature = "trace_task_details")]
         span.record("immutable", is_immutable || is_now_immutable);
 
         if !queue.is_empty() || !old_edges.is_empty() {
@@ -2120,19 +2150,33 @@ impl<B: BackingStorage> TurboTasksBackendInner<B> {
 
         let mut queue = AggregationUpdateQueue::new();
         for dependent_task_id in output_dependent_tasks {
+            #[cfg(feature = "trace_task_output_dependencies")]
+            let span = tracing::trace_span!(
+                "invalidate output dependency",
+                task = %task_id,
+                dependent_task = %dependent_task_id,
+                result = tracing::field::Empty,
+            )
+            .entered();
             if ctx.is_once_task(dependent_task_id) {
                 // once tasks are never invalidated
+                #[cfg(feature = "trace_task_output_dependencies")]
+                span.record("result", "once task");
                 continue;
             }
             let dependent = ctx.task(dependent_task_id, TaskDataCategory::All);
             if dependent.has_key(&CachedDataItemKey::OutdatedOutputDependency { target: task_id }) {
                 // output dependency is outdated, so it hasn't read the output yet
                 // and doesn't need to be invalidated
+                #[cfg(feature = "trace_task_output_dependencies")]
+                span.record("result", "outdated dependency");
                 continue;
             }
             if !dependent.has_key(&CachedDataItemKey::OutputDependency { target: task_id }) {
                 // output dependency has been removed, so the task doesn't depend on the
                 // output anymore and doesn't need to be invalidated
+                #[cfg(feature = "trace_task_output_dependencies")]
+                span.record("result", "no backward dependency");
                 continue;
             }
             make_task_dirty_internal(
@@ -2144,6 +2188,8 @@ impl<B: BackingStorage> TurboTasksBackendInner<B> {
                 &mut queue,
                 ctx,
             );
+            #[cfg(feature = "trace_task_output_dependencies")]
+            span.record("result", "marked dirty");
         }
 
         queue.execute(ctx);

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

@@ -182,6 +182,13 @@ impl Operation for CleanupOldEdgesOperation {
                                 }
                             }
                             OutdatedEdge::OutputDependency(output_task_id) => {
+                                #[cfg(feature = "trace_task_output_dependencies")]
+                                let _span = tracing::trace_span!(
+                                    "remove output dependency",
+                                    task = %output_task_id,
+                                    dependent_task = %task_id
+                                )
+                                .entered();
                                 {
                                     let mut task = ctx.task(output_task_id, TaskDataCategory::Data);
                                     task.remove(&CachedDataItemKey::OutputDependent {

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

@@ -224,6 +224,7 @@ pub fn make_task_dirty_internal(
         #[cfg(feature = "trace_task_dirty")]
         let _span = tracing::trace_span!(
             "make task stale",
+            task_id = display(task_id),
             name = ctx.get_task_description(task_id),
             cause = %TaskDirtyCauseInContext::new(&cause, ctx)
         )
@@ -244,6 +245,7 @@ pub fn make_task_dirty_internal(
             #[cfg(feature = "trace_task_dirty")]
             let _span = tracing::trace_span!(
                 "task already dirty",
+                task_id = display(task_id),
                 name = ctx.get_task_description(task_id),
                 cause = %TaskDirtyCauseInContext::new(&cause, ctx)
             )
@@ -275,6 +277,7 @@ pub fn make_task_dirty_internal(
     #[cfg(feature = "trace_task_dirty")]
     let _span = tracing::trace_span!(
         "make task dirty",
+        task_id = display(task_id),
         name = ctx.get_task_description(task_id),
         cause = %TaskDirtyCauseInContext::new(&cause, ctx)
     )

turbopack/crates/turbo-tasks/Cargo.toml

@@ -12,6 +12,7 @@ bench = false
 default = []
 tokio_tracing = ["tokio/tracing"]
 hanging_detection = []
+task_id_details = []
 
 # 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

@@ -427,6 +427,8 @@ pub struct TurboTasksError {
 #[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
 pub struct TurboTaskContextError {
     pub task: RcStr,
+    #[cfg(feature = "task_id_details")]
+    pub task_id: Option<TaskId>,
     pub source: Option<TurboTasksExecutionError>,
 }
 
@@ -438,9 +440,11 @@ pub enum TurboTasksExecutionError {
 }
 
 impl TurboTasksExecutionError {
-    pub fn with_task_context(&self, task: impl Display) -> Self {
+    pub fn with_task_context(&self, task: impl Display, _task_id: Option<TaskId>) -> Self {
         TurboTasksExecutionError::TaskContext(Arc::new(TurboTaskContextError {
             task: RcStr::from(task.to_string()),
+            #[cfg(feature = "task_id_details")]
+            task_id: _task_id,
             source: Some(self.clone()),
         }))
     }
@@ -468,6 +472,14 @@ impl Display for TurboTasksExecutionError {
                 write!(f, "{}", error.message)
             }
             TurboTasksExecutionError::TaskContext(context_error) => {
+                #[cfg(feature = "task_id_details")]
+                if let Some(task_id) = context_error.task_id {
+                    return write!(
+                        f,
+                        "Execution of {} ({}) failed",
+                        context_error.task, task_id
+                    );
+                }
                 write!(f, "Execution of {} failed", context_error.task)
             }
         }

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

@@ -281,6 +281,15 @@ pub enum TaskPersistence {
     Transient,
 }
 
+impl Display for TaskPersistence {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        match self {
+            TaskPersistence::Persistent => write!(f, "persistent"),
+            TaskPersistence::Transient => write!(f, "transient"),
+        }
+    }
+}
+
 #[derive(Clone, Copy, Debug, Eq, PartialEq, Default)]
 pub enum ReadConsistency {
     /// The default behavior for most APIs. Reads are faster, but may return stale values, which
@@ -826,7 +835,7 @@ impl<B: Backend + 'static> TurboTasks<B> {
                 let output = match result {
                     Ok(raw_vc) => OutputContent::Link(raw_vc),
                     Err(err) => OutputContent::Error(
-                        TurboTasksExecutionError::from(err).with_task_context(task_type),
+                        TurboTasksExecutionError::from(err).with_task_context(task_type, None),
                     ),
                 };
 

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

@@ -103,6 +103,22 @@ impl Debug for RawVc {
     }
 }
 
+impl Display for RawVc {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        match self {
+            RawVc::TaskOutput(task_id) => write!(f, "output of task {}", **task_id),
+            RawVc::TaskCell(task_id, cell_id) => {
+                write!(f, "{} of task {}", cell_id, **task_id)
+            }
+            RawVc::LocalOutput(execution_id, local_task_id, task_persistence) => write!(
+                f,
+                "output of local task {} ({}, {})",
+                **local_task_id, **execution_id, task_persistence
+            ),
+        }
+    }
+}
+
 impl RawVc {
     pub fn is_resolved(&self) -> bool {
         match self {