Repository: vercel/next.js
File: turbopack/crates/turbo-tasks-backend/src/backend/mod.rs (Rust)

Code before:
        drop(task);
    }

    fn run_backend_job<'a>(
        self: &'a Arc<Self>,
        job: TurboTasksBackendJob,
        turbo_tasks: &'a dyn TurboTasksBackendApi<TurboTasksBackend<B>>,
    ) -> Pin<Box<dyn Future<Output = ()> + Send + 'a>> {
        Box::pin(async move {
            match job {
                TurboTasksBackendJob::InitialSnapshot | TurboTasksBackendJob::FollowUpSnapshot => {
                    debug_assert!(self.should_persist());

                    let last_snapshot = self.last_snapshot.load(Ordering::Relaxed);
                    let mut last_snapshot = self.start_time + Duration::from_millis(last_snapshot);
                    let mut idle_start_listener = self.idle_start_event.listen();
                    let mut idle_end_listener = self.idle_end_event.listen();
                    let mut fresh_idle = true;
                    loop {
                        const FIRST_SNAPSHOT_WAIT: Duration = Duration::from_secs(300);
                        const SNAPSHOT_INTERVAL: Duration = Duration::from_secs(120);
                        let idle_timeout = *IDLE_TIMEOUT;
                        let (time, mut reason) =
                            if matches!(job, TurboTasksBackendJob::InitialSnapshot) {
                                (FIRST_SNAPSHOT_WAIT, "initial snapshot timeout")
                            } else {
                                (SNAPSHOT_INTERVAL, "regular snapshot interval")
                            };

                        let until = last_snapshot + time;
                        if until > Instant::now() {
                            let mut stop_listener = self.stopping_event.listen();
                            if self.stopping.load(Ordering::Acquire) {
                                return;
                            }
                            let mut idle_time = if turbo_tasks.is_idle() && fresh_idle {
                                Instant::now() + idle_timeout
                            } else {
                                far_future()
                            };
                            loop {
                                tokio::select! {
                                    _ = &mut stop_listener => {
                                        return;
                                    },
                                    _ = &mut idle_start_listener => {
                                        fresh_idle = true;
                                        idle_time = Instant::now() + idle_timeout;
                                        idle_start_listener = self.idle_start_event.listen()
                                    },
                                    _ = &mut idle_end_listener => {

Review comment (suggestion):
so if it isn't caused by eventual consistency we shouldn't worry too much about not invalidating more downstream tasks because the _user_ should be addressing the issue?

im trying to think through the worst case scenario here and i guess it would involve us just leaking memory in error scenarios, which isn't too concerning i guess.

Code after:
            drop(task);

            // Notify dependent tasks that are waiting for this task to finish
            done_event.notify(usize::MAX);
        }

        drop(old_content);

        if let Some(data_update) = data_update {
            AggregationUpdateQueue::run(data_update, ctx);
        }

        reschedule
    }

    fn task_execution_completed_cleanup(
        &self,
        ctx: &mut impl ExecuteContext<'_>,
        task_id: TaskId,
        cell_counters: &AutoMap<ValueTypeId, u32, BuildHasherDefault<FxHasher>, 8>,
        is_error: bool,
        removed_data: &mut Vec<CachedDataItem>,
    ) {
        let mut task = ctx.task(task_id, TaskDataCategory::All);

        // An error is potentially caused by a eventual consistency, so we avoid updating cells
        // after an error as it is likely transient and we want to keep the dependent tasks
        // clean to avoid re-executions.
        if !is_error {
            // Remove no longer existing cells and
            // find all outdated data items (removed cells, outdated edges)
            // Note: For persistent tasks we only want to call extract_if when there are actual
            // cells to remove to avoid tracking that as modification.
            // Note: We do not mark the tasks as dirty here, as these tasks are unused or stale
            // anyway and we want to avoid needless re-executions. When the cells become
            // used again, they are invalidated from the update cell operation.
            let is_cell_unused = |cell: CellId| {
                cell_counters
                    .get(&cell.type_id)
                    .is_none_or(|start_index| cell.index >= *start_index)
            };
            if task_id.is_transient()
                || iter_many!(task, CellData { cell } => cell).any(is_cell_unused)
            {
                removed_data.extend(task.extract_if(CachedDataItemType::CellData, |key, _| {
                    matches!(key, CachedDataItemKey::CellData { cell } if cell_counters.get(&cell.type_id).is_none_or(|start_index| cell.index >= *start_index))
                }));
            }
            if task_id.is_transient()
                || iter_many!(task, TransientCellData { cell } => cell).any(is_cell_unused)
            {