vercel/next.js · #98614
turbo-tasks-gc: Invalidating a deleted task should be a no-op
turbopack/crates/turbo-tasks-backend/src/backend/operation/invalidate.rs24 + / 1 −
@@ -60,7 +60,7 @@ impl Operation for InvalidateOperation { } => { let mut queue = AggregationUpdateQueue::new(); for task_id in task_ids {- make_task_dirty(+ try_make_task_dirty( task_id, #[cfg(feature = "task_dirty_cause")] cause.clone(),@@ -88,6 +88,7 @@ impl Operation for InvalidateOperation { } } +/// Marks a task dirty. The task must exist. pub fn make_task_dirty( task_id: TaskId, #[cfg(feature = "task_dirty_cause")] cause: TaskDirtyCause,@@ -105,6 +106,28 @@ pub fn make_task_dirty( ); } +/// Marks a task dirty, doing nothing if it no longer exists.+///+/// Intended for invalidation usecases.+fn try_make_task_dirty(+ task_id: TaskId,+ #[cfg(feature = "task_dirty_cause")] cause: TaskDirtyCause,+ queue: &mut AggregationUpdateQueue,+ ctx: &mut impl ExecuteContext<'_>,+) {+ let Some(mut task) = ctx.try_get_task(task_id, TaskDataCategory::All) else {+ return;+ };+ make_task_dirty_internal(+ &mut task,+ true,+ #[cfg(feature = "task_dirty_cause")]+ cause,+ queue,+ ctx,+ );+}+ /// Requires the guard to be allocated with [TaskDataCategory::All] pub fn make_task_dirty_internal<'e, E: ExecuteContext<'e>>( task: &mut E::TaskGuardImpl,turbopack/crates/turbo-tasks-backend/src/backend/operation/mod.rs208 + / 113 −
@@ -9,6 +9,7 @@ mod update_cell; mod update_collectible; use std::{ fmt::{Debug, Display, Formatter},+ ops::{Deref, DerefMut}, sync::Arc, }; @@ -30,7 +31,7 @@ use crate::{ EventDescription, TaskDataCategory, TurboTasksBackend, cell_data::CellData, snapshot_coordinator::{OperationGuard, SnapshotPhase},- storage::{SpecificTaskDataCategory, StorageWriteGuard, TrackOutcome},+ storage::{SpecificTaskDataCategory, StorageWriteGuard, TaskEntryGuard, TrackOutcome}, storage_schema::{TaskStorage, TaskStorageAccessors}, }, data::{ActivenessState, CollectibleRef, Dirtyness, InProgressState, TransientTask},@@ -40,9 +41,44 @@ pub trait Operation: Encode + Decode<()> + Default + TryFrom<AnyOperation, Error fn execute(self, ctx: &mut impl ExecuteContext<'_>); } -/// Whether an [`ExecuteContext`] task open may create the task or requires it to already exist.-/// A private impl detail behind the two public methods ([`ExecuteContext::task`] = `MustExist`,-/// [`ExecuteContext::open_or_create_task_storage`] = `MaybeCreate`).+/// The task storage `open_task` is working with, which may or may not still own its map entry.+enum OpenedTask<'a> {+ Owned(TaskEntryGuard<'a>),+ Restored(StorageWriteGuard<'a>),+}++impl<'a> OpenedTask<'a> {+ fn into_write_guard(self) -> StorageWriteGuard<'a> {+ match self {+ OpenedTask::Owned(g) => g.into_write_guard(),+ OpenedTask::Restored(g) => g,+ }+ }+}++impl Deref for OpenedTask<'_> {+ type Target = TaskStorage;+ fn deref(&self) -> &Self::Target {+ match self {+ OpenedTask::Owned(g) => g,+ OpenedTask::Restored(g) => g,+ }+ }+}++impl DerefMut for OpenedTask<'_> {+ fn deref_mut(&mut self) -> &mut Self::Target {+ match self {+ OpenedTask::Owned(g) => g,+ OpenedTask::Restored(g) => g,+ }+ }+}++/// Whether an [`ExecuteContext`] task open may create the task, requires it to already exist, or+/// tolerates its absence. A private impl detail behind the three public methods+/// ([`ExecuteContext::task`] = `MustExist`, [`ExecuteContext::open_or_create_task_storage`] =+/// `MaybeCreate`, [`ExecuteContext::try_get_task`] = `AllowMissing`). #[derive(Copy, Clone, Debug, PartialEq, Eq)] enum TaskAccess { /// Open the task, creating it if it does not exist: `access_mut` inserts a blank entry, then@@ -52,8 +88,10 @@ enum TaskAccess { /// task that exists in neither memory nor persistent storage is a bug — a stale reference to an /// already-collected or never-created task — and this refuses to fabricate a blank for it. ///- /// This is very much expression a 'foreign key constraint' on the database.+ /// This is very much expressing a 'foreign key constraint' on the database. MustExist,+ /// Open a task that may legitimately be gone or `deleted`.+ AllowMissing, } // TODO: consider removing this trait (and `TaskGuard`) in favor of the concrete types. Each has@@ -73,6 +111,15 @@ pub trait ExecuteContext<'e>: Sized { /// The check applies only to persistent tasks; a `MustExist` open of a transient id falls /// through to create. See `ExecuteContextImpl::open_task`. fn task(&mut self, task_id: TaskId, category: TaskDataCategory) -> Self::TaskGuardImpl;+ /// Opens a task that may legitimately be gone, returning `None` if it is.+ ///+ /// Gone covers both a task that exists nowhere and one that is soft-deleted: the caller cannot+ /// tell those apart, since only the timing of the next eviction separates them.+ fn try_get_task(+ &mut self,+ task_id: TaskId,+ category: TaskDataCategory,+ ) -> Option<Self::TaskGuardImpl>; /// Opens a task, materializing an in-memory storage entry for it if one is not resident yet /// (inserting a blank, then restoring `category` from disk if present). Use only where the /// task's storage may not be resident: the first connect of a freshly-minted child (threads can@@ -299,26 +346,24 @@ impl<'e> ExecuteContextImpl<'e> { task_id: TaskId, category: TaskDataCategory, access: TaskAccess,- ) -> TaskGuardImpl<'e> {+ ) -> Option<TaskGuardImpl<'e>> { self.task_lock_counter.acquire(); - // A resident entry always corresponds to a task that exists (only a `MaybeCreate` open ever- // inserts a blank, and only for a task being created). A `MustExist` open therefore only- // needs to prove existence when the entry looks like a fresh blank: nothing restored, not a- // new task. (A fully-evicted resident task also matches this shape, but it is on disk, so- // the `found_on_disk` check below clears it — the panic fires only when the task is in- // neither memory nor disk.)- let mut task = self.backend.storage.access_mut(task_id);- // The `MustExist` non-fabrication check applies only to **persistent** tasks: they have- // disk backing and are the subject of the stale-reference/GC concern. A transient task has- // no disk copy and is materialized lazily in memory (a strongly-consistent read can open a- // transient root through the aggregation graph before its storage entry exists), so a- // `MustExist` open of a transient id is a no-op that falls through to create.- let maybe_fabricated = access == TaskAccess::MustExist- && !task_id.is_transient()- && !task.flags.is_restored(TaskDataCategory::Meta)- && !task.flags.is_restored(TaskDataCategory::Data)- && !task.flags.new_task();+ let mut task = OpenedTask::Owned(self.backend.storage.access_entry_mut(task_id));+ // Treat deleted tasks under Allowmissing as missing+ if access == TaskAccess::AllowMissing && task.flags.deleted() {+ self.task_lock_counter.release();+ return None;+ }++ // IF the caller cares about existence (either to panic or return None), check if this is an+ // effectively blank task+ let needs_existence_check =+ matches!(access, TaskAccess::MustExist | TaskAccess::AllowMissing)+ && !task_id.is_transient()+ && !task.flags.is_restored(TaskDataCategory::Meta)+ && !task.flags.is_restored(TaskDataCategory::Data)+ && !task.flags.new_task(); if !task.flags.is_restored(category) { if task_id.is_transient() { task.flags.set_restored(TaskDataCategory::All);@@ -343,103 +388,144 @@ impl<'e> ExecuteContextImpl<'e> { task.flags.set_meta_restoring(true); } - if do_data || do_meta || data_restoring || meta_restoring {- let waiting_for_restore = data_restoring || meta_restoring;- if waiting_for_restore {- // The caller holds the task id outside the graph while waiting, so pin it- // against GC until the restored guard reaches the use boundary. Eviction is- // still allowed; the wait loop restores the category again if needed.- task.update_and_get_transient_ref_count(1);- }- // Drop lock while doing I/O (our I/O can overlap with the other thread).- drop(task);-- // Perform I/O for categories we claimed.- let storage_data = do_data- .then(|| self.restore_task_data(task_id, SpecificTaskDataCategory::Data));- let storage_meta = do_meta- .then(|| self.restore_task_data(task_id, SpecificTaskDataCategory::Meta));-- // Whether our own I/O found the task on disk (in any restored category).- // Another thread restoring it concurrently (`*_restoring`)- // also proves existence: it only sets the restoring bit- // after finding the task.- let found_on_disk = restored_from_disk(&storage_data)- || restored_from_disk(&storage_meta)- || data_restoring- || meta_restoring;-- // Wait for categories claimed by another thread (after our I/O).- // Reuse the returned write guard to avoid a second lock acquisition.- task = if let Some(cat) = wait_category(data_restoring, meta_restoring) {- self.wait_for_restore_or_panic(task_id, cat)- } else {- self.backend.storage.access_mut(task_id)- };- if waiting_for_restore {- // This caller owns the pin and releases it only after acquiring the task- // guard it is about to use.- task.update_and_get_transient_ref_count(-1);- }+ // `!is_restored(category)` above already implies this: for a single category+ // it is the same predicate, and for `All` both are the disjunction+ // `!data_restored || !meta_restored`. Asserted so that a future change to+ // `is_restored` or `TaskDataCategory` surfaces here instead of silently+ // skipping the restore.+ debug_assert!(+ needs_data || needs_meta,+ "task({task_id}, {category:?}): not restored, yet neither category needs \+ restoring"+ ); - // Apply results and clear restoring bits.- if let Some(result) = storage_data- && let Err(e) =- apply_restore_result(&mut task, result, SpecificTaskDataCategory::Data)- {- drop(task);- self.backend.storage.restored.notify(usize::MAX);- panic!("Failed to restore data for task {task_id}: {e:?}");- }- if let Some(result) = storage_meta- && let Err(e) =- apply_restore_result(&mut task, result, SpecificTaskDataCategory::Meta)- {- drop(task);- self.backend.storage.restored.notify(usize::MAX);- panic!("Failed to restore meta for task {task_id}: {e:?}");- }+ let waiting_for_restore = data_restoring || meta_restoring;+ if waiting_for_restore {+ // The caller holds the task id outside the graph while waiting, so pin it+ // against GC until the restored guard reaches the use boundary. Eviction is+ // still allowed; the wait loop restores the category again if needed.+ task.update_and_get_transient_ref_count(1);+ }+ // Drop lock while doing I/O (our I/O can overlap with the other thread).+ drop(task); - if do_data || do_meta {- // Keep the guard through return. Once the restoring bit is clear, eviction- // may otherwise drop the category before this caller can use it.- self.backend.storage.restored.notify(usize::MAX);- }+ // Perform I/O for categories we claimed.+ let storage_data = do_data+ .then(|| self.restore_task_data(task_id, SpecificTaskDataCategory::Data));+ let storage_meta = do_meta+ .then(|| self.restore_task_data(task_id, SpecificTaskDataCategory::Meta)); - // The caller asserted this task exists (`MustExist`), but it looked like a- // fresh blank and restore found nothing on disk (and no one- // else was restoring it): it exists nowhere. Fail loudly- // rather than hand back a fabricated task, which- // would silently corrupt the graph. (The leftover blank entry is inert; the- // panic tears the process down.)- //- // This also fires if the on-disk cache is corrupt or truncated. That is the- // intended behavior: there is no recovery path for reading a cell on a task- // that is missing from disk, and the panic is self-healing — the cache is- // discarded and rebuilt on the next run.- assert!(- !(maybe_fabricated && !found_on_disk),- "task({task_id}, MustExist): task exists in neither memory nor persistent \- storage — a stale reference to an already-collected or never-created task"- );+ // Whether our own I/O found the task on disk (in any restored category).+ //+ // A category another thread claimed (`*_restoring`) counts as found. The bit is+ // set before that thread's I/O, so on its own it proves only that someone is+ // looking — but this thread waits for that restore below, and a peer that comes up+ // empty clears the category's `restored` bit before returning, so the blank cannot+ // be mistaken for a real task.+ let found_on_disk = restored_from_disk(&storage_data)+ || restored_from_disk(&storage_meta)+ || data_restoring+ || meta_restoring;++ // Wait for categories claimed by another thread (after our I/O).+ // Reuse the returned write guard to avoid a second lock acquisition.+ // Reuse the guard the waiter already holds; only our own-I/O path can come+ // up empty and need to discard the entry.+ task = if let Some(cat) = wait_category(data_restoring, meta_restoring) {+ OpenedTask::Restored(self.wait_for_restore_or_panic(task_id, cat)) } else {- // Nothing to restore (no categories claimed, none in progress) yet the entry- // looked like a fresh blank for a task asserted to exist: it does not exist.- assert!(- !maybe_fabricated,+ OpenedTask::Owned(self.backend.storage.access_entry_mut(task_id))+ };+ if waiting_for_restore {+ // This caller owns the pin and releases it only after acquiring the task+ // guard it is about to use.+ task.update_and_get_transient_ref_count(-1);+ }++ // Apply results and clear restoring bits.+ if let Some(result) = storage_data+ && let Err(e) =+ apply_restore_result(&mut task, result, SpecificTaskDataCategory::Data)+ {+ drop(task);+ self.backend.storage.restored.notify(usize::MAX);+ panic!("Failed to restore data for task {task_id}: {e:?}");+ }+ if let Some(result) = storage_meta+ && let Err(e) =+ apply_restore_result(&mut task, result, SpecificTaskDataCategory::Meta)+ {+ drop(task);+ self.backend.storage.restored.notify(usize::MAX);+ panic!("Failed to restore meta for task {task_id}: {e:?}");+ }++ if do_data || do_meta {+ // Keep the guard through return. Once the restoring bit is clear, eviction+ // may otherwise drop the category before this caller can use it.+ self.backend.storage.restored.notify(usize::MAX);+ }++ // It looked like a fresh blank and restore found nothing on disk (and no one+ // else was restoring it): it exists nowhere. An `AllowMissing` open reports+ // that; a `MustExist` open fails loudly rather than hand+ // back a fabricated task, which would silently corrupt the+ // graph. (The leftover blank entry is inert; the+ // panic tears the process down.)+ if needs_existence_check && !found_on_disk {+ if access == TaskAccess::AllowMissing {+ // If the count is non-zero that means another thread is mid-restore or+ // otherwise connecting to it. If they are waiting with AllowMissing they+ // will perform the discard and if not then they want the blank and that is+ // also fine.+ if task.gc_transient_ref_count() == 0 {+ match task {+ OpenedTask::Owned(g) => g.discard(),+ OpenedTask::Restored(_) => {+ unreachable!(+ "a task restored by another thread exists and is never \+ discarded"+ )+ }+ };+ } else {+ // A waiter pins the entry, so it cannot be removed here. Reset the+ // categories we read back to "never looked" instead so the other reader+ // re-reads and also observes absence+ //+ // Only the categories this thread claimed are ours to clear. One a+ // peer restored is the peer's to report, and it reaches this same+ // code to clear its own.+ debug_assert!(+ !(do_data && task.flags.data_restoring())+ && !(do_meta && task.flags.meta_restoring()),+ "task({task_id}): apply_restore_result should have cleared the \+ restoring bits for the categories we claimed"+ );+ if do_data {+ task.flags.set_data_restored(false);+ }+ if do_meta {+ task.flags.set_meta_restored(false);+ }+ }+ self.task_lock_counter.release();+ return None;+ }+ panic!( "task({task_id}, MustExist): task exists in neither memory nor persistent \ storage — a stale reference to an already-collected or never-created task" ); } } }- TaskGuardImpl {- task,+ Some(TaskGuardImpl {+ task: task.into_write_guard(), task_id, #[cfg(debug_assertions)] category, task_lock_counter: self.task_lock_counter.clone(),- }+ }) } /// Restores one category for a task from persistent storage. `None` means the task was **not@@ -974,7 +1060,7 @@ fn wait_category(wait_data: bool, wait_meta: bool) -> Option<TaskDataCategory> { /// the restored flag. On error, returns the error so the caller can drop the task lock, /// notify waiters, and panic. fn apply_restore_result(- task: &mut StorageWriteGuard<'_>,+ task: &mut (impl DerefMut<Target = TaskStorage> + ?Sized), result: Result<Option<TaskStorage>>, category: SpecificTaskDataCategory, ) -> Result<()> {@@ -1018,6 +1104,15 @@ impl<'e> ExecuteContext<'e> for ExecuteContextImpl<'e> { fn task(&mut self, task_id: TaskId, category: TaskDataCategory) -> Self::TaskGuardImpl { self.open_task(task_id, category, TaskAccess::MustExist)+ .expect("a MustExist open either yields a task or panics")+ }++ fn try_get_task(+ &mut self,+ task_id: TaskId,+ category: TaskDataCategory,+ ) -> Option<Self::TaskGuardImpl> {+ self.open_task(task_id, category, TaskAccess::AllowMissing) } fn open_or_create_task_storage(@@ -1026,6 +1121,7 @@ impl<'e> ExecuteContext<'e> for ExecuteContextImpl<'e> { category: TaskDataCategory, ) -> Self::TaskGuardImpl { self.open_task(task_id, category, TaskAccess::MaybeCreate)+ .expect("a MaybeCreate open always yields a task") } fn prepare_tasks(@@ -1080,11 +1176,11 @@ impl<'e> ExecuteContext<'e> for ExecuteContextImpl<'e> { // in memory and has no disk copy): a task that looks like a freshly-inserted blank (nothing // restored, not a new task) and that restore does not find on disk exists nowhere — a stale // reference. See `TaskAccess::MustExist`.- let maybe_fabricated1 = !task_id1.is_transient()+ let needs_existence_check1 = !task_id1.is_transient() && !task1.flags.is_restored(TaskDataCategory::Meta) && !task1.flags.is_restored(TaskDataCategory::Data) && !task1.flags.new_task();- let maybe_fabricated2 = !task_id2.is_transient()+ let needs_existence_check2 = !task_id2.is_transient() && !task2.flags.is_restored(TaskDataCategory::Meta) && !task2.flags.is_restored(TaskDataCategory::Data) && !task2.flags.new_task();@@ -1232,15 +1328,15 @@ impl<'e> ExecuteContext<'e> for ExecuteContextImpl<'e> { // A `MustExist` pair open must not fabricate: a task that looked like a fresh blank and // was not found on disk exists nowhere (a stale reference). See // `TaskAccess::MustExist`. Only reachable in the restore branch — a task- // already resident/restored (the else path) has `maybe_fabricated ==+ // already resident/restored (the else path) has `needs_existence_check == // false`. assert!(- !(maybe_fabricated1 && !found_on_disk1),+ !(needs_existence_check1 && !found_on_disk1), "task_pair({task_id1}, .., MustExist): task exists in neither memory nor \ persistent storage — a stale reference to a never-created task" ); assert!(- !(maybe_fabricated2 && !found_on_disk2),+ !(needs_existence_check2 && !found_on_disk2), "task_pair(.., {task_id2}, MustExist): task exists in neither memory nor \ persistent storage — a stale reference to a never-created task" );@@ -1490,7 +1586,6 @@ pub trait TaskGuard: Debug + TaskStorageAccessors { /// /// How much this proves depends on the guard's category — with only `Meta` open it is a sound /// pre-filter that cannot see dependency edges, and with `All` open it is authoritative. See- /// [`TaskStorage::gc_maybe_collectible`] for the full contract. fn is_gc_collectible(&self) -> bool { // Transient-ness is a property of the id, not the storage; transient tasks are never // collected.turbopack/crates/turbo-tasks-backend/src/backend/storage.rs52 + / 0 −
@@ -522,6 +522,20 @@ impl Storage { } } + /// Like [`Self::access_mut`], but keeps the map entry so the caller can still remove it.+ pub fn access_entry_mut(&self, key: TaskId) -> TaskEntryGuard<'_> {+ let entry = match self.map.entry(key) {+ dashmap::mapref::entry::Entry::Occupied(e) => e,+ dashmap::mapref::entry::Entry::Vacant(e) => {+ e.insert_entry(Box::new(TaskStorage::new()))+ }+ };+ TaskEntryGuard {+ storage: self,+ entry,+ }+ }+ /// Read-only access to an already resident task. Returns `None` if the task isnt in memory /// resident. The closure runs while a shard read lock is held, so it must be cheap and must /// not re-enter the map.@@ -831,6 +845,44 @@ impl Storage { } } +/// A write guard that still owns its map entry, so the task can be removed under the lock that is+/// already held.+///+/// Use [`Storage::access_entry_mut`] to obtain one. Convert it with [`Self::into_write_guard`] once+/// removal is no longer a possibility, or call [`Self::discard`] to drop the entry outright.+pub struct TaskEntryGuard<'a> {+ storage: &'a Storage,+ entry: dashmap::mapref::entry::OccupiedEntry<'a, TaskId, Box<TaskStorage>>,+}++impl<'a> TaskEntryGuard<'a> {+ /// Removes this task's entry.+ pub fn discard(self) {+ self.entry.remove();+ }++ /// Gives up the ability to remove the entry, yielding an ordinary write guard.+ pub fn into_write_guard(self) -> StorageWriteGuard<'a> {+ StorageWriteGuard {+ storage: self.storage,+ inner: self.entry.into_ref().into(),+ }+ }+}++impl Deref for TaskEntryGuard<'_> {+ type Target = TaskStorage;+ fn deref(&self) -> &Self::Target {+ self.entry.get()+ }+}++impl DerefMut for TaskEntryGuard<'_> {+ fn deref_mut(&mut self) -> &mut Self::Target {+ self.entry.get_mut()+ }+}+ pub struct StorageWriteGuard<'a> { storage: &'a Storage, inner: RefMut<'a, TaskId, Box<TaskStorage>>,turbopack/crates/turbo-tasks-backend/tests/gc_collection.rs79 + / 3 −
@@ -11,7 +11,7 @@ use anyhow::Result; use turbo_tasks::{ResolvedVc, TaskId, Vc, prevent_gc}; use crate::{- gc_fixture::{Selector, create_selector},+ gc_fixture::{Constant, Selector, create_constant, create_selector}, util::create_tt, }; @@ -47,6 +47,29 @@ async fn branch_b() -> Result<Vc<u32>> { Ok(Vc::cell(2 + *leaf(20).await?)) } +/// Reads `observed` so the task registers an invalidator on it, then goes out of the live graph+/// when the selector flips. Whether that invalidator outlives the task is the point of+/// `mutating_a_state_read_by_a_collected_task_does_not_panic`.+#[turbo_tasks::function]+async fn state_reader(observed: ResolvedVc<Constant>) -> Result<Vc<u32>> {+ Ok(Vc::cell(*observed.await?.get()))+}++/// Reads `state_reader` only while the selector is false, so flipping it disconnects the reader.+#[turbo_tasks::function(operation, root)]+async fn select_state_reader(+ selector: ResolvedVc<Selector>,+ observed: ResolvedVc<Constant>,+) -> Result<Vc<u32>> {+ let use_b = *selector.await?.get();+ let value = if use_b {+ *branch_b().await?+ } else {+ *state_reader(*observed).await?+ };+ Ok(Vc::cell(value))+}+ /// A task that pins itself against GC while executing. Once pinned it must survive collection even /// after it is disconnected. #[turbo_tasks::function]@@ -114,7 +137,6 @@ async fn gc_collects_disconnected_subtree() { ); // Flipping back must recompute branch_a fresh, since it was collected.- let tt3 = tt.clone(); let result = turbo_tasks::run_once(tt.clone(), async move { let selector_op = create_selector(true); let selector_vc = selector_op.resolve().strongly_consistent().await?;@@ -123,7 +145,6 @@ async fn gc_collects_disconnected_subtree() { assert_eq!(*output.read_strongly_consistent().await?, 22); selector.set(false); assert_eq!(*output.read_strongly_consistent().await?, 11);- let _ = &tt3; anyhow::Ok(()) }) .await;@@ -266,3 +287,58 @@ async fn unpin_after_stop_does_not_panic() { tt.unpin_task_for_gc(leaf_id); }++/// A `State` keeps an `Invalidator` for every task that read it, and those entries are plain task+/// ids with nothing keeping the task alive. Collecting a reader therefore leaves a dangling+/// invalidator behind, this test ensures that that doesn't cause a panic.+#[tokio::test(flavor = "multi_thread", worker_threads = 2)]+async fn mutating_a_state_read_by_a_collected_task_does_not_panic() {+ let (tt, _persistence_dir) =+ create_tt("mutating_a_state_read_by_a_collected_task_does_not_panic");+ let tt2 = tt.clone();++ let result = turbo_tasks::run_once(tt.clone(), async move {+ let selector_op = create_selector(false);+ let selector_vc = selector_op.resolve().strongly_consistent().await?;+ let selector = selector_op.read_strongly_consistent().await?;+ let observed_vc = create_constant().resolve().strongly_consistent().await?;++ // `state_reader` reads `observed`, registering an invalidator on that State.+ let output = select_state_reader(selector_vc, observed_vc);+ output.read_strongly_consistent().await?;++ // Flip so `state_reader` leaves the live graph; its invalidator stays on `observed`.+ selector.set(true);+ output.read_strongly_consistent().await?;++ anyhow::Ok(())+ })+ .await;+ result.unwrap();++ let collected = tt2.backend().gc_for_testing(&tt2);+ assert!(+ collected > 0,+ "the disconnected state reader should have been collected"+ );++ // Required, and not just for realism: the evict is what selects the code path under test.+ // GC only soft-deletes, leaving the task resident, and a weak open of a resident-but-deleted+ // task returns early on the `deleted` flag. Evicting drops it from memory (and tombstones it on+ // disk), so the open below instead reaches the exists-nowhere case in+ // `ExecuteContextImpl::open_task` -- nothing restored, nothing found on disk -- which is the+ // one that would panic under `MustExist`. Drop this line and the test still passes, but it+ // stops covering that path.+ tt2.backend().snapshot_and_evict_for_testing(&tt2);++ // Mutating the State now walks its invalidator list, which still names the collected reader.+ let result = turbo_tasks::run_once(tt.clone(), async move {+ let observed = create_constant().read_strongly_consistent().await?;+ observed.set(1);+ anyhow::Ok(())+ })+ .await;+ result.unwrap();++ tt.stop_and_wait().await;+}turbopack/crates/turbo-tasks-backend/tests/gc_resurrection.rs3 + / 13 −
@@ -230,7 +230,6 @@ async fn gc_resurrect_on_reconnect() { // Reconnect the subtree (selector back to false) BEFORE any snapshot. Reading `reader` again // connects it, which must resurrect it (and its leaves, as it re-reads them).- let tt3 = tt.clone(); let result = turbo_tasks::run_once(tt.clone(), async move { let selector_op = create_selector(false); let selector_vc = selector_op.resolve().strongly_consistent().await?;@@ -244,23 +243,20 @@ async fn gc_resurrect_on_reconnect() { expected, "resurrected reader must recompute the correct value" );- let _ = &tt3; anyhow::Ok(()) }) .await; result.unwrap(); // A snapshot+evict now must NOT have tombstoned/hard-deleted the resurrected subtree. tt2.backend().snapshot_and_evict_for_testing(&tt2);- let tt4 = tt.clone(); let result = turbo_tasks::run_once(tt.clone(), async move { let selector_op = create_selector(false); let selector_vc = selector_op.resolve().strongly_consistent().await?; let constant_op = create_constant(); let constant_vc = constant_op.resolve().strongly_consistent().await?; let output = select_reader(selector_vc, constant_vc); assert_eq!(*output.read_strongly_consistent().await?, expected);- let _ = &tt4; anyhow::Ok(()) }) .await;@@ -319,7 +315,6 @@ async fn select_imm_reader(selector: ResolvedVc<Selector>) -> Result<Vc<u32>> { #[tokio::test(flavor = "multi_thread", worker_threads = 4)] async fn gc_resurrect_immutable_recomputes() { let (tt, _persistence_dir) = create_tt("gc_resurrect_immutable_recomputes");- let tt2 = tt.clone(); let expected: u32 = (0..IMM_FANOUT).fold(0u32, |a, b| a.wrapping_add(b * 3)); let result = turbo_tasks::run_once(tt.clone(), async move {@@ -349,7 +344,7 @@ async fn gc_resurrect_immutable_recomputes() { // Collect the disconnected subtree. The entries stay resident (no snapshot yet), so the tasks // are soft-deleted rather than gone.- let collected = tt2.backend().gc_for_testing(&tt2);+ let collected = tt.backend().gc_for_testing(&tt); assert_eq!( collected, IMM_FANOUT as usize + 1,@@ -360,10 +355,8 @@ async fn gc_resurrect_immutable_recomputes() { // This must recompute it rather than serve a stale value. Done before the reconnect below, // while the subtree is still collected — afterwards the leaves are live again and a read would // legitimately hit a fresh cell, proving nothing.- let tt_direct = tt.clone(); let result = turbo_tasks::run_once(tt.clone(), async move { assert_eq!(*read_imm_leaf(0).read_strongly_consistent().await?, 0);- let _ = &tt_direct; anyhow::Ok(()) }) .await;@@ -377,7 +370,6 @@ async fn gc_resurrect_immutable_recomputes() { // Reconnect BEFORE any snapshot. These tasks were never persisted, so there is nothing on disk // to restore — the only way back to a correct value is re-execution.- let tt3 = tt.clone(); let result = turbo_tasks::run_once(tt.clone(), async move { let selector_op = create_selector(false); let selector_vc = selector_op.resolve().strongly_consistent().await?;@@ -389,7 +381,7 @@ async fn gc_resurrect_immutable_recomputes() { expected, "a resurrected immutable task must recompute the correct value" );- let _ = &tt3;+ anyhow::Ok(()) }) .await;@@ -404,14 +396,12 @@ async fn gc_resurrect_immutable_recomputes() { // A snapshot + evict must not have tombstoned the resurrected subtree, and the restored data // must survive the round trip.- tt2.backend().snapshot_and_evict_for_testing(&tt2);- let tt4 = tt.clone();+ tt.backend().snapshot_and_evict_for_testing(&tt); let result = turbo_tasks::run_once(tt.clone(), async move { let selector_op = create_selector(false); let selector_vc = selector_op.resolve().strongly_consistent().await?; let output = select_imm_reader(selector_vc); assert_eq!(*output.read_strongly_consistent().await?, expected);- let _ = &tt4; anyhow::Ok(()) }) .await;turbopack/crates/turbo-tasks-backend/tests/gc_stress.rs1 + / 4 −
@@ -35,7 +35,6 @@ async fn gc_re_rooting_stays_flat() { tt: &Arc<TurboTasks<TurboTasksBackend>>, gen_value: u32, ) -> (usize, usize, bool) {- let tt_inner = tt.clone(); turbo_tasks::run_once(tt.clone(), async move { // `create_generation` is a cached root operation, so this returns the same task each // round.@@ -47,7 +46,6 @@ async fn gc_re_rooting_stays_flat() { } let output = wide_root(generation_vc, WIDTH); output.read_strongly_consistent().await?;- let _ = &tt_inner; anyhow::Ok(()) }) .await@@ -119,14 +117,13 @@ async fn gc_re_rooting_stays_flat() { ); // The live graph must still compute correctly after all the churn.- let tt3 = tt.clone(); let result = turbo_tasks::run_once(tt.clone(), async move { let generation_op = create_generation(); let generation_vc = generation_op.resolve().strongly_consistent().await?; let output = wide_root(generation_vc, WIDTH); let expected: u32 = expected_value(ROUNDS, WIDTH); assert_eq!(*output.read_strongly_consistent().await?, expected);- let _ = &tt3;+ anyhow::Ok(()) }) .await;