fix collectibles counting and aggregation

vercel · Nov 12, 2024 · e2d793a · e2d793a
1 parent d81a9f7
commit e2d793a
Show file tree

Hide file tree

Showing 4 changed files with 48 additions and 20 deletions.
diff --git a/turbopack/crates/turbo-tasks-backend/src/backend/mod.rs b/turbopack/crates/turbo-tasks-backend/src/backend/mod.rs
@@ -1281,6 +1281,15 @@ impl<B: BackingStorage> TurboTasksBackendInner<B> {
                         _ => None,
                     },
                 ));
+                old_edges.extend(task.iter(CachedDataItemIndex::Collectibles).filter_map(
+                    |(key, value)| match (key, value) {
+                        (
+                            CachedDataItemKey::OutdatedCollectible { collectible },
+                            CachedDataItemValue::OutdatedCollectible { value },
+                        ) => Some(OutdatedEdge::Collectible(*collectible, *value)),
+                        _ => None,
+                    },
+                ));
             }
             if self.should_track_dependencies() {
                 old_edges.extend(task.iter(CachedDataItemIndex::Dependencies).filter_map(

diff --git a/turbopack/crates/turbo-tasks-backend/src/backend/operation/aggregation_update.rs b/turbopack/crates/turbo-tasks-backend/src/backend/operation/aggregation_update.rs
@@ -146,7 +146,7 @@ impl AggregatedDataUpdate {
         let aggregation = get_aggregation_number(task);
         let mut dirty_container_count = Default::default();
         let mut collectibles_update: Vec<_> =
-            get_many!(task, Collectible { collectible } => (*collectible, 1));
+            get_many!(task, Collectible { collectible } count if *count > 0 => (*collectible, 1));
         if is_aggregating_node(aggregation) {
             dirty_container_count = get!(task, AggregatedDirtyContainerCount)
                 .cloned()

diff --git a/turbopack/crates/turbo-tasks-backend/src/backend/operation/cleanup_old_edges.rs b/turbopack/crates/turbo-tasks-backend/src/backend/operation/cleanup_old_edges.rs
@@ -99,7 +99,7 @@ impl Operation for CleanupOldEdgesOperation {
                             }
                             OutdatedEdge::Collectible(collectible, count) => {
                                 let mut collectibles = Vec::new();
-                                collectibles.push((collectible, count));
+                                collectibles.push((collectible, -count));
                                 outdated.retain(|e| match e {
                                     OutdatedEdge::Collectible(collectible, count) => {
                                         collectibles.push((*collectible, -*count));
@@ -108,9 +108,24 @@ impl Operation for CleanupOldEdgesOperation {
                                     _ => true,
                                 });
                                 let mut task = ctx.task(task_id, TaskDataCategory::All);
-                                for &(collectible, count) in collectibles.iter() {
-                                    update_count!(task, Collectible { collectible }, -count);
+                                for (collectible, count) in collectibles.iter_mut() {
+                                    if update_count!(
+                                        task,
+                                        Collectible {
+                                            collectible: *collectible
+                                        },
+                                        *count
+                                    ) {
+                                        if *count > 0 {
+                                            *count = 1;
+                                        } else {
+                                            *count = -1;
+                                        }
+                                    } else {
+                                        *count = 0;
+                                    }
                                 }
+                                collectibles.retain(|(_, count)| *count != 0);
                                 queue.extend(AggregationUpdateJob::data_update(
                                     &mut task,
                                     AggregatedDataUpdate::new().collectibles_update(collectibles),

diff --git a/turbopack/crates/turbo-tasks-backend/src/backend/operation/update_collectible.rs b/turbopack/crates/turbo-tasks-backend/src/backend/operation/update_collectible.rs
@@ -20,7 +20,7 @@ impl UpdateCollectibleOperation {
     pub fn run(
         task_id: TaskId,
         collectible: CollectibleRef,
-        count: i32,
+        mut count: i32,
         mut ctx: impl ExecuteContext,
     ) {
         if !ctx.should_track_children() {
@@ -32,26 +32,30 @@ impl UpdateCollectibleOperation {
         let outdated = get!(task, OutdatedCollectible { collectible }).copied();
         if let Some(outdated) = outdated {
             if count > 0 && outdated > 0 {
-                update_count!(
-                    task,
-                    OutdatedCollectible { collectible },
-                    -min(count, outdated)
-                );
+                let shared = min(count, outdated);
+                update_count!(task, OutdatedCollectible { collectible }, -shared);
+                count -= shared;
             } else if count < 0 && outdated < 0 {
-                update_count!(
-                    task,
-                    OutdatedCollectible { collectible },
-                    min(-count, -outdated)
-                );
+                let shared = min(-count, -outdated);
+                update_count!(task, OutdatedCollectible { collectible }, shared);
+                count += shared;
             } else {
                 // Not reduced from outdated
             }
         }
-        update_count!(task, Collectible { collectible }, count);
-        queue.extend(AggregationUpdateJob::data_update(
-            &mut task,
-            AggregatedDataUpdate::new().collectibles_update(vec![(collectible, count)]),
-        ));
+        if count != 0 && update_count!(task, Collectible { collectible }, count) {
+            if count > 0 {
+                queue.extend(AggregationUpdateJob::data_update(
+                    &mut task,
+                    AggregatedDataUpdate::new().collectibles_update(vec![(collectible, 1)]),
+                ));
+            } else {
+                queue.extend(AggregationUpdateJob::data_update(
+                    &mut task,
+                    AggregatedDataUpdate::new().collectibles_update(vec![(collectible, -1)]),
+                ));
+            }
+        }
 
         drop(task);