fix aggregation of outdated children and collectibles (#6885)

### Description

The delayed removal of children and collectibles for "in progress" tasks
of the aggregate tree causes some inconsistency in the aggregated tree
when "in progress" task are connected to other node. We need to fix that
by reporting outdated children and collectibles as children/collectibles
of the "in progress" tasks.

This fixes an race condition when temporary errors are incorrectly shows
as final result. e. g. it was visible as `can't resolve
"ACTIONS_MODULE9"` in next.js.


Closes PACK-2192

crates/turbo-tasks-memory/src/task.rs

@@ -1418,6 +1418,15 @@ impl Task {
                 thresholds_job = ensure_thresholds(&aggregation_context, &mut guard);
                 let TaskGuard { guard, .. } = guard;
                 let mut state = TaskMetaStateWriteGuard::full_from(guard.into_inner(), self);
+                if let TaskStateType::InProgress {
+                    outdated_children, ..
+                } = &mut state.state_type
+                {
+                    if outdated_children.remove(&child_id) {
+                        state.children.insert(child_id);
+                        return;
+                    }
+                }
                 if state.children.insert(child_id) {
                     add_job = Some(
                         state

crates/turbo-tasks-memory/src/task/aggregation.rs

@@ -415,17 +415,34 @@ impl<'l> AggregationItemLock for TaskGuard<'l> {
 
     fn number_of_children(&self) -> usize {
         match self.guard {
-            TaskMetaStateWriteGuard::Full(ref guard) => guard.children.len(),
+            TaskMetaStateWriteGuard::Full(ref guard) => match &guard.state_type {
+                TaskStateType::InProgress {
+                    outdated_children, ..
+                } => guard.children.len() + outdated_children.len(),
+                _ => guard.children.len(),
+            },
             TaskMetaStateWriteGuard::Partial(_) | TaskMetaStateWriteGuard::Unloaded(_) => 0,
         }
     }
 
     fn children(&self) -> Self::ChildrenIter<'_> {
         match self.guard {
             TaskMetaStateWriteGuard::Full(ref guard) => {
-                Some(guard.children.iter().map(Cow::Borrowed))
-                    .into_iter()
-                    .flatten()
+                let outdated_children = match &guard.state_type {
+                    TaskStateType::InProgress {
+                        outdated_children, ..
+                    } => Some(outdated_children.iter().map(Cow::Borrowed)),
+                    _ => None,
+                };
+                Some(
+                    guard
+                        .children
+                        .iter()
+                        .map(Cow::Borrowed)
+                        .chain(outdated_children.into_iter().flatten()),
+                )
+                .into_iter()
+                .flatten()
             }
             TaskMetaStateWriteGuard::Partial(_) | TaskMetaStateWriteGuard::Unloaded(_) => {
                 None.into_iter().flatten()
@@ -457,6 +474,17 @@ impl<'l> AggregationItemLock for TaskGuard<'l> {
                         change.collectibles.push((trait_type_id, collectible, 1));
                     }
                 }
+                if let TaskStateType::InProgress {
+                    outdated_collectibles,
+                    ..
+                } = &guard.state_type
+                {
+                    if let Some(collectibles) = outdated_collectibles.as_ref() {
+                        for (&(trait_type_id, collectible), _) in collectibles.iter() {
+                            change.collectibles.push((trait_type_id, collectible, 1));
+                        }
+                    }
+                }
                 if change.is_empty() {
                     None
                 } else {
@@ -491,6 +519,17 @@ impl<'l> AggregationItemLock for TaskGuard<'l> {
                         change.collectibles.push((trait_type_id, collectible, -1));
                     }
                 }
+                if let TaskStateType::InProgress {
+                    outdated_collectibles,
+                    ..
+                } = &guard.state_type
+                {
+                    if let Some(collectibles) = outdated_collectibles.as_ref() {
+                        for (&(trait_type_id, collectible), _) in collectibles.iter() {
+                            change.collectibles.push((trait_type_id, collectible, -1));
+                        }
+                    }
+                }
                 if change.is_empty() {
                     None
                 } else {