turbo-tasks-backend: batch schedule dirty tasks in aggregation_update (#91461)

### What?

Batch the scheduling of dirty tasks during aggregation update processing
instead of scheduling them one at a time.

In `aggregation_update.rs`, `find_and_schedule_dirty_internal`
previously called `ctx.schedule()` immediately for each task it decided
to schedule. This involved a separate `ctx.task(id,
TaskDataCategory::All)` call per task to read `leaf_distance` and
`has_output` before calling `turbo_tasks.schedule(id, priority)`.

### Why?

Each individual `ctx.task(...)` call may require restoring task data
from persistent storage (disk I/O). When many tasks need scheduling —
e.g. during a large invalidation wave, or initially when all
session-dependent tasks are scheduled — doing this one-at-a-time is
inefficient.

`for_each_task_all` (via `prepare_tasks_with_callback`) can batch the
restoration of persistent task data, amortizing I/O overhead across all
tasks in the batch.

### How?

- Added a `scheduled_tasks: FxHashMap<TaskId, TaskPriority>` field to
`AggregationUpdateQueue` (skipped in bincode serialization — transient
state that doesn't survive suspend/resume).
- `find_and_schedule_dirty_internal` now inserts `(task_id,
parent_priority)` into `scheduled_tasks` instead of calling
`ctx.schedule()` directly.
- At the end of `process()` (in the final `else` branch, when all
sub-queues are exhausted), if `scheduled_tasks` is non-empty,
`ctx.for_each_task_all` is called to batch-fetch the tasks and schedule
them, then `scheduled_tasks` is cleared. This maximizes batch size by
deferring all scheduling to the end of the operation.
- `scheduled_tasks` is included in `is_empty()` (with exhaustive struct
destructuring) so the compiler enforces the check stays up to date.

---------

Co-authored-by: Tobias Koppers <[email protected]>
Co-authored-by: Claude <[email protected]>

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

@@ -24,7 +24,8 @@ use tracing::span::Span;
 ))]
 use tracing::trace_span;
 use turbo_tasks::{
-    FxIndexMap, TaskExecutionReason, TaskId, backend::CachedTaskType, event::EventDescription,
+    FxIndexMap, TaskExecutionReason, TaskId, TaskPriority, backend::CachedTaskType,
+    event::EventDescription,
 };
 
 #[cfg(feature = "trace_task_dirty")]
@@ -861,6 +862,8 @@ pub struct AggregationUpdateQueue {
     balance_queue: FxRingSet<BalanceJob>,
     #[bincode(with = "turbo_bincode::ringset")]
     optimize_queue: FxRingSet<OptimizeJob>,
+    #[bincode(skip, default = "FxHashMap::default")]
+    scheduled_tasks: FxHashMap<TaskId, TaskPriority>,
 }
 
 impl AggregationUpdateQueue {
@@ -873,6 +876,7 @@ impl AggregationUpdateQueue {
             find_and_schedule: FxRingSet::default(),
             balance_queue: FxRingSet::default(),
             optimize_queue: FxRingSet::default(),
+            scheduled_tasks: FxHashMap::default(),
         }
     }
 
@@ -885,12 +889,14 @@ impl AggregationUpdateQueue {
             balance_queue,
             optimize_queue,
             done_aggregation_number_updates: _,
+            scheduled_tasks,
         } = self;
         jobs.is_empty()
             && aggregation_number_updates.is_empty()
             && find_and_schedule.is_empty()
             && balance_queue.is_empty()
             && optimize_queue.is_empty()
+            && scheduled_tasks.is_empty()
     }
 
     /// Pushes a job to the queue.
@@ -1249,6 +1255,13 @@ impl AggregationUpdateQueue {
                 self.find_and_schedule.drain(..count).collect();
             self.find_and_schedule_dirty(jobs, ctx);
             false
+        } else if !self.scheduled_tasks.is_empty() {
+            ctx.for_each_task_all(self.scheduled_tasks.keys().copied(), |task, ctx| {
+                let parent_priority = self.scheduled_tasks[&task.id()];
+                ctx.schedule_task(task, parent_priority);
+            });
+            self.scheduled_tasks.clear();
+            false
         } else {
             true
         }
@@ -1482,7 +1495,7 @@ impl AggregationUpdateQueue {
             let description = EventDescription::new(|| task.get_task_desc_fn());
             if task.add_scheduled(reason, description) {
                 drop(task);
-                ctx.schedule(task_id, parent_priority);
+                self.scheduled_tasks.insert(task_id, parent_priority);
             }
         }
     }