diff --git a/turbopack/crates/turbo-tasks-backend/src/backend/mod.rs b/turbopack/crates/turbo-tasks-backend/src/backend/mod.rs index 13b8f1935724..a24db97efaf6 100644 --- a/turbopack/crates/turbo-tasks-backend/src/backend/mod.rs +++ b/turbopack/crates/turbo-tasks-backend/src/backend/mod.rs @@ -330,6 +330,20 @@ impl TurboTasksBackend { (had_new_data, counts) } + /// Opens `task` with the must-exist [`ExecuteContext::task`] and drops the guard. Test-only + /// hook to exercise the non-fabricating existence guarantee: this panics (debug builds) if + /// `task` exists in neither memory nor persistent storage (rather than fabricating a + /// blank). + #[doc(hidden)] + pub fn assert_task_exists_for_testing( + &self, + task: TaskId, + turbo_tasks: &TurboTasks, + ) { + let mut ctx = self.execute_context(turbo_tasks); + let _ = ctx.task(task, TaskDataCategory::All); + } + fn should_restore(&self) -> bool { self.options.storage_mode.is_some() } @@ -1244,7 +1258,7 @@ impl TurboTasksBackend { None }; - SnapshotItem { + SnapshotItem::Put { task_id, meta, data, @@ -1793,7 +1807,9 @@ impl TurboTasksBackend { turbo_tasks: &TurboTasks, ) -> String { let mut ctx = self.execute_context(turbo_tasks); - let task = ctx.task(task_id, TaskDataCategory::Data); + // Diagnostic path: the caller may name any id, including one that no longer exists, so this + // must not assert existence. A nonexistent task falls through to the "unknown" case below. + let task = ctx.open_or_create_task_storage(task_id, TaskDataCategory::Data); if let Some(value) = task.get_persistent_task_type() { value.to_string() } else if let Some(value) = task.get_transient_task_type() { diff --git a/turbopack/crates/turbo-tasks-backend/src/backend/operation/connect_child.rs b/turbopack/crates/turbo-tasks-backend/src/backend/operation/connect_child.rs index e13866f2d8cd..359c9e30cbaf 100644 --- a/turbopack/crates/turbo-tasks-backend/src/backend/operation/connect_child.rs +++ b/turbopack/crates/turbo-tasks-backend/src/backend/operation/connect_child.rs @@ -79,7 +79,10 @@ impl ConnectChildOperation { task: child_task_id, }); } else { - let mut child_task = ctx.task(child_task_id, TaskDataCategory::Meta); + // First connect of this child: its id is minted but the storage entry may not exist + // yet, and concurrent connects race to be the one that first touches it. + let mut child_task = + ctx.open_or_create_task_storage(child_task_id, TaskDataCategory::Meta); let has_output = child_task.has_output(); // An already constructed top-level task was made a root when it was first connected. // It may still be dirty and need to run; this only avoids repeating the idempotent diff --git a/turbopack/crates/turbo-tasks-backend/src/backend/operation/mod.rs b/turbopack/crates/turbo-tasks-backend/src/backend/operation/mod.rs index 9b2c1a2abe23..fb3ee5847c0d 100644 --- a/turbopack/crates/turbo-tasks-backend/src/backend/operation/mod.rs +++ b/turbopack/crates/turbo-tasks-backend/src/backend/operation/mod.rs @@ -38,12 +38,49 @@ pub trait Operation: Encode + Decode<()> + Default + TryFrom); } +/// 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`). +#[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 + /// the requested category is restored from disk (staying empty if there is nothing on disk). + MaybeCreate, + /// Open a task the caller expects to **already exist** (resident, or restorable from disk). A + /// 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. + MustExist, +} + pub trait ExecuteContext<'e>: Sized { type TaskGuardImpl: TaskGuard + 'e; fn child_context<'l, 'r>(&'r self) -> impl ChildExecuteContext<'l> + use<'e, 'l, Self> where 'e: 'l; + /// Opens a task that must **already exist**, restoring the requested `category` if needed. A + /// task that exists in neither memory nor persistent storage is a stale reference, so this + /// panics rather than fabricate a blank. This is the common case; use + /// [`Self::open_or_create_task_storage`] only where the task may be getting materialized for + /// the first time. + /// + /// 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, 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 + /// race to first-touch it). + /// + /// This creates *storage for* an already-minted `TaskId`; it does not mint one. Compare + /// `TurboTasksBackend::get_or_create_task`, which takes a function and arguments and returns a + /// new `TaskId`. + fn open_or_create_task_storage( + &mut self, + task_id: TaskId, + category: TaskDataCategory, + ) -> Self::TaskGuardImpl; /// Prepares (as in fetches from persistent storage) a list of tasks. /// The iterator should not have duplicates, as this would cause over-fetching. fn prepare_tasks( @@ -81,6 +118,9 @@ pub trait ExecuteContext<'e>: Sized { func, ) } + /// Opens two tasks that must **already exist** under a single lock acquisition (to atomically + /// read/mutate an edge between them). Both ids are opened `MustExist` — an edge only exists + /// between already-materialized tasks. fn task_pair( &mut self, task_id1: TaskId, @@ -183,21 +223,160 @@ impl<'e> ExecuteContextImpl<'e> { } } + fn open_task( + &mut self, + task_id: TaskId, + category: TaskDataCategory, + access: TaskAccess, + ) -> 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(); + if !task.flags.is_restored(category) { + if task_id.is_transient() { + task.flags.set_restored(TaskDataCategory::All); + } else { + // Collect which categories need restoring while we have the lock + let needs_data = + category.includes_data() && !task.flags.is_restored(TaskDataCategory::Data); + let needs_meta = + category.includes_meta() && !task.flags.is_restored(TaskDataCategory::Meta); + + // Check whether another thread is currently restoring each category. + let data_restoring = needs_data && task.flags.data_restoring(); + let meta_restoring = needs_meta && task.flags.meta_restoring(); + + // Claim categories no one else is restoring. + let do_data = needs_data && !data_restoring; + let do_meta = needs_meta && !meta_restoring; + if do_data { + task.flags.set_data_restoring(true); + } + if do_meta { + task.flags.set_meta_restoring(true); + } + + if do_data || do_meta || data_restoring || meta_restoring { + // 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) + }; + + // 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 { + // Drop the lock before notifying so woken threads don't + // immediately contend on the same DashMap shard. + drop(task); + self.backend.storage.restored.notify(usize::MAX); + task = self.backend.storage.access_mut(task_id); + } + + // 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" + ); + } 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, + "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, + 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 + /// present** on disk. A `MaybeCreate` open treats that the same as empty storage; a `MustExist` + /// open uses it to refuse to fabricate a task that exists nowhere. fn restore_task_data( &self, task_id: TaskId, category: SpecificTaskDataCategory, - ) -> Result { + ) -> Result> { debug_assert!( self.backend.should_restore(), "restore_task_data called when should_restore() is false" ); - let mut storage = TaskStorage::default(); self.backend .backing_storage - .lookup_data(task_id, category, &mut storage) - .with_context(|| format!("Failed to restore {category:?} for {task_id}"))?; - Ok(storage) + .lookup_data(task_id, category) + .with_context(|| format!("Failed to restore {category:?} for {task_id}")) } fn restore_task_data_batch( @@ -447,7 +626,7 @@ impl<'e> ExecuteContextImpl<'e> { Ok(data) => { for (item, &idx) in data.into_iter().zip(&tasks_to_restore_for_data_indices) { - tasks[idx].data_restore_result = Some(Ok(item)); + tasks[idx].data_restore_result = Some(Ok(Some(item))); } } Err(e) => { @@ -479,7 +658,7 @@ impl<'e> ExecuteContextImpl<'e> { Ok(data) => { for (item, &idx) in data.into_iter().zip(&tasks_to_restore_for_meta_indices) { - tasks[idx].meta_restore_result = Some(Ok(item)); + tasks[idx].meta_restore_result = Some(Ok(Some(item))); } } Err(e) => { @@ -595,10 +774,12 @@ impl<'e> ExecuteContextImpl<'e> { struct TaskRestoreEntry { task_id: TaskId, category: TaskDataCategory, - /// Result of restoring the data category (set in Phase 1b, consumed in Phase 1c). - data_restore_result: Option>, - /// Result of restoring the meta category (set in Phase 1b, consumed in Phase 1c). - meta_restore_result: Option>, + /// Result of restoring the data category (set in Phase 1b, consumed in Phase 1c). The inner + /// `Option` is `None` when the task was not present on disk. + data_restore_result: Option>>, + /// Result of restoring the meta category (set in Phase 1b, consumed in Phase 1c). The inner + /// `Option` is `None` when the task was not present on disk. + meta_restore_result: Option>>, /// Another thread claimed the data restore; we must wait in Phase 3. wait_data: bool, /// Another thread claimed the meta restore; we must wait in Phase 3. @@ -609,6 +790,12 @@ struct TaskRestoreEntry { self_restored: bool, } +/// Whether a restore we performed proves the task exists on disk: we ran the I/O (outer `Some`), it +/// succeeded (`Ok`), and it found a key (inner `Some`). A failed or skipped restore proves nothing. +fn restored_from_disk(result: &Option>>) -> bool { + matches!(result, Some(Ok(Some(_)))) +} + /// Combines per-category booleans into a single `TaskDataCategory` for waiting. fn wait_category(wait_data: bool, wait_meta: bool) -> Option { match (wait_data, wait_meta) { @@ -627,12 +814,15 @@ fn wait_category(wait_data: bool, wait_meta: bool) -> Option { /// notify waiters, and panic. fn apply_restore_result( task: &mut StorageWriteGuard<'_>, - result: Result, + result: Result>, category: SpecificTaskDataCategory, ) -> Result<()> { let task_category = TaskDataCategory::from(category); match result { + // A task absent from disk applies as empty storage; only a `MustExist` open treats absence + // as an error, and it checks that before getting here. Ok(storage) => { + let storage = storage.unwrap_or_default(); if task.flags.is_restored(task_category) { // Already restored by another path (e.g., initialize_new_task racing // with our I/O). Just clear the restoring bit so waiting threads @@ -666,86 +856,15 @@ impl<'e> ExecuteContext<'e> for ExecuteContextImpl<'e> { } fn task(&mut self, task_id: TaskId, category: TaskDataCategory) -> Self::TaskGuardImpl { - self.task_lock_counter.acquire(); - - let mut task = self.backend.storage.access_mut(task_id); - if !task.flags.is_restored(category) { - if task_id.is_transient() { - task.flags.set_restored(TaskDataCategory::All); - } else { - // Collect which categories need restoring while we have the lock - let needs_data = - category.includes_data() && !task.flags.is_restored(TaskDataCategory::Data); - let needs_meta = - category.includes_meta() && !task.flags.is_restored(TaskDataCategory::Meta); - - // Check whether another thread is currently restoring each category. - let data_restoring = needs_data && task.flags.data_restoring(); - let meta_restoring = needs_meta && task.flags.meta_restoring(); - - // Claim categories no one else is restoring. - let do_data = needs_data && !data_restoring; - let do_meta = needs_meta && !meta_restoring; - if do_data { - task.flags.set_data_restoring(true); - } - if do_meta { - task.flags.set_meta_restoring(true); - } - - if do_data || do_meta || data_restoring || meta_restoring { - // 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)); - - // 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) - }; - - // 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:?}"); - } + self.open_task(task_id, category, TaskAccess::MustExist) + } - if do_data || do_meta { - // Drop the lock before notifying so woken threads don't - // immediately contend on the same DashMap shard. - drop(task); - self.backend.storage.restored.notify(usize::MAX); - task = self.backend.storage.access_mut(task_id); - } - } - } - } - TaskGuardImpl { - task, - task_id, - #[cfg(debug_assertions)] - category, - task_lock_counter: self.task_lock_counter.clone(), - } + fn open_or_create_task_storage( + &mut self, + task_id: TaskId, + category: TaskDataCategory, + ) -> Self::TaskGuardImpl { + self.open_task(task_id, category, TaskAccess::MaybeCreate) } fn prepare_tasks( @@ -795,6 +914,20 @@ impl<'e> ExecuteContext<'e> for ExecuteContextImpl<'e> { let (mut task1, mut task2) = self.backend.storage.access_pair_mut(task_id1, task_id2); + // `task_pair` is always a `MustExist` open (both endpoints of an existing edge). Existence + // check mirroring `open_task` (persistent tasks only — a transient task materializes lazily + // 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() + && !task1.flags.is_restored(TaskDataCategory::Meta) + && !task1.flags.is_restored(TaskDataCategory::Data) + && !task1.flags.new_task(); + let maybe_fabricated2 = !task_id2.is_transient() + && !task2.flags.is_restored(TaskDataCategory::Meta) + && !task2.flags.is_restored(TaskDataCategory::Data) + && !task2.flags.new_task(); + // Collect what needs restoring for each task. let needs_data1 = category.includes_data() && !task1.flags.is_restored(TaskDataCategory::Data); @@ -852,6 +985,17 @@ impl<'e> ExecuteContext<'e> for ExecuteContextImpl<'e> { let storage_meta2 = do_meta2.then(|| self.restore_task_data(task_id2, SpecificTaskDataCategory::Meta)); + // Whether our own I/O found each task on disk (any restored category). A concurrent + // restorer (`*_restoring`) also proves existence. + let found_on_disk1 = restored_from_disk(&storage_data1) + || restored_from_disk(&storage_meta1) + || data1_restoring + || meta1_restoring; + let found_on_disk2 = restored_from_disk(&storage_data2) + || restored_from_disk(&storage_meta2) + || data2_restoring + || meta2_restoring; + // Wait for categories claimed by another thread (after our I/O, so they can overlap). // Returns write guards; drop them since we re-acquire via access_pair_mut below. if let Some(cat) = wait_category(data1_restoring, meta1_restoring) { @@ -914,6 +1058,22 @@ impl<'e> ExecuteContext<'e> for ExecuteContextImpl<'e> { task1 = t1; task2 = t2; } + + // 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 == + // false`. + assert!( + !(maybe_fabricated1 && !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), + "task_pair(.., {task_id2}, MustExist): task exists in neither memory nor \ + persistent storage — a stale reference to a never-created task" + ); } ( diff --git a/turbopack/crates/turbo-tasks-backend/src/backend/storage.rs b/turbopack/crates/turbo-tasks-backend/src/backend/storage.rs index ea75ee4efe46..2e6cfc9944dd 100644 --- a/turbopack/crates/turbo-tasks-backend/src/backend/storage.rs +++ b/turbopack/crates/turbo-tasks-backend/src/backend/storage.rs @@ -1044,7 +1044,7 @@ mod tests { _: &super::TaskStorage, _: &mut TurboBincodeBuffer, ) -> SnapshotItem { - SnapshotItem { + SnapshotItem::Put { task_id, meta: Some(TurboBincodeBuffer::default()), data: None, @@ -1112,7 +1112,7 @@ mod tests { // The pre-snapshot snapshot copy should have been encoded and returned. assert_eq!(items.len(), 1); - assert_eq!(items[0].task_id, task_id); + assert_eq!(items[0].task_id(), task_id); { let guard = storage.access_mut(task_id); @@ -1179,7 +1179,7 @@ mod tests { .collect(); assert_eq!(items.len(), 1); - assert_eq!(items[0].task_id, task_id); + assert_eq!(items[0].task_id(), task_id); { let guard = storage.access_mut(task_id); @@ -1227,7 +1227,7 @@ mod tests { .collect(); assert_eq!(items.len(), 1); - assert_eq!(items[0].task_id, task_id); + assert_eq!(items[0].task_id(), task_id); // The entry must be gone from the map now that it has been persisted. assert!( @@ -1324,7 +1324,7 @@ mod tests { .flat_map(|shard| shard.into_iter()) .collect(); assert_eq!(items.len(), 1); - assert_eq!(items[0].task_id, modified_id); + assert_eq!(items[0].task_id(), modified_id); } #[tokio::test(flavor = "multi_thread")] diff --git a/turbopack/crates/turbo-tasks-backend/src/backing_storage.rs b/turbopack/crates/turbo-tasks-backend/src/backing_storage.rs index 7c4a75fe161c..655915be55a0 100644 --- a/turbopack/crates/turbo-tasks-backend/src/backing_storage.rs +++ b/turbopack/crates/turbo-tasks-backend/src/backing_storage.rs @@ -8,15 +8,40 @@ use turbo_tasks_hash::Xxh3Hash64Hasher; pub type TaskTypeHash = [u8; 8]; -/// A single item yielded by the snapshot iterator during persistence. -pub struct SnapshotItem { - pub task_id: TaskId, - /// Serialized task meta data, if modified - pub meta: Option, - /// Serialized task data, if modified - pub data: Option, - /// Task type for new tasks that need to be added to the task cache - pub task_type_hash: Option, +/// A single item yielded by the snapshot iterator during persistence: either a put (persist a +/// modified task's meta/data + optionally register a new task's type) or a delete (tombstone a +/// GC-collected task's on-disk copy). Both ride the one iterator `save_snapshot` consumes, so +/// tombstones are applied in the same commit and batch as the puts. +pub enum SnapshotItem { + Put { + task_id: TaskId, + /// Serialized task meta data, if modified + meta: Option, + /// Serialized task data, if modified + data: Option, + /// Task type for new tasks that need to be added to the task cache + task_type_hash: Option, + }, + // Constructed by the GC pass that emits `Delete` for soft-deleted tasks, which lands in a + // later PR in the stack. + #[allow(dead_code)] + Delete { + task_id: TaskId, + /// The deleted task's `TaskCache` key. Always present: only persistent tasks are + /// collected, and those always have a task type. + task_type_hash: TaskTypeHash, + }, +} + +impl SnapshotItem { + /// The task this item persists or tombstones. (Currently only used by tests, which assert on + /// the id of items yielded by the snapshot iterator.) + #[cfg(test)] + pub fn task_id(&self) -> TaskId { + match self { + SnapshotItem::Put { task_id, .. } | SnapshotItem::Delete { task_id, .. } => *task_id, + } + } } /// Computes a deterministic 64-bit hash of a CachedTaskType for use as a TaskCache key. diff --git a/turbopack/crates/turbo-tasks-backend/src/database/turbo/mod.rs b/turbopack/crates/turbo-tasks-backend/src/database/turbo/mod.rs index 32c9c304a87a..7f7603ec7731 100644 --- a/turbopack/crates/turbo-tasks-backend/src/database/turbo/mod.rs +++ b/turbopack/crates/turbo-tasks-backend/src/database/turbo/mod.rs @@ -234,6 +234,25 @@ impl<'a> TurboWriteBatch<'a> { .put(key_space as u32, key.into_static(), value.into()) } + /// Writes a delete (tombstone) for `key` into the write batch. + /// + /// Use [`Self::delete_value`] to remove a single mapping from a MultiValue KeySpace + pub fn delete(&self, key_space: KeySpace, key: WriteBuffer<'_>) -> Result<()> { + self.batch.delete(key_space as u32, key.into_static()) + } + + /// Writes a tombstone for a single `key` -> `value` mapping, leaving other values under `key` + /// intact. Only valid for `MultiValue` families (`TaskCache`). + pub fn delete_value( + &self, + key_space: KeySpace, + key: WriteBuffer<'_>, + value: WriteBuffer<'_>, + ) -> Result<()> { + self.batch + .delete_value(key_space as u32, key.into_static(), value.into()) + } + /// Flushes a key space of the write batch, reducing the amount of buffered memory used. /// Does not commit any data persistently. /// diff --git a/turbopack/crates/turbo-tasks-backend/src/kv_backing_storage.rs b/turbopack/crates/turbo-tasks-backend/src/kv_backing_storage.rs index 50d1e8fd5f9d..d08e6759fb09 100644 --- a/turbopack/crates/turbo-tasks-backend/src/kv_backing_storage.rs +++ b/turbopack/crates/turbo-tasks-backend/src/kv_backing_storage.rs @@ -236,55 +236,74 @@ impl TurboBackingStorage { let batch = self.inner.database.write_batch()?; { - let _span = tracing::trace_span!("update task data").entered(); + let span = tracing::trace_span!("update task data"); let mut snapshot_meta = parallel::map_collect_owned::<_, _, Result>>(snapshots, |shard: I| { + let _span = span.clone().entered(); let mut max_new_task_id = 0; let mut data_items = 0; let mut meta_items = 0; let mut task_cache_items = 0; - for SnapshotItem { - task_id, - meta, - data, - task_type_hash, - } in shard - { - let key = IntKey::new(*task_id); - let key = key.as_ref(); - if let Some(meta) = meta { - batch.put( - KeySpace::TaskMeta, - WriteBuffer::Borrowed(key), - WriteBuffer::SmallVec(meta), - )?; - meta_items += 1; - } - if let Some(data) = data { - batch.put( - KeySpace::TaskData, - WriteBuffer::Borrowed(key), - WriteBuffer::SmallVec(data), - )?; - data_items += 1; - } - // Write task cache entry inline if this is a new task - if let Some(task_type_hash) = task_type_hash { - batch.put( - KeySpace::TaskCache, - WriteBuffer::Borrowed(&task_type_hash), - WriteBuffer::Borrowed(key), - )?; - task_cache_items += 1; - max_new_task_id = max_new_task_id.max(*task_id); + for item in shard { + match item { + SnapshotItem::Put { + task_id, + meta, + data, + task_type_hash, + } => { + let key = IntKey::new(*task_id); + let key = key.as_ref(); + if let Some(meta) = meta { + batch.put( + KeySpace::TaskMeta, + WriteBuffer::Borrowed(key), + WriteBuffer::SmallVec(meta), + )?; + meta_items += 1; + } + if let Some(data) = data { + batch.put( + KeySpace::TaskData, + WriteBuffer::Borrowed(key), + WriteBuffer::SmallVec(data), + )?; + data_items += 1; + } + // Register the task type only for new tasks. + if let Some(task_type_hash) = task_type_hash { + batch.put( + KeySpace::TaskCache, + WriteBuffer::Borrowed(&task_type_hash), + WriteBuffer::Borrowed(key), + )?; + task_cache_items += 1; + max_new_task_id = max_new_task_id.max(*task_id); + } + } + SnapshotItem::Delete { + task_id, + task_type_hash, + } => { + let key = IntKey::new(*task_id); + let key = key.as_ref(); + batch.delete(KeySpace::TaskMeta, WriteBuffer::Borrowed(key))?; + batch.delete(KeySpace::TaskData, WriteBuffer::Borrowed(key))?; + // TaskCache is MultiValue, delete just this id from the bucket. + batch.delete_value( + KeySpace::TaskCache, + WriteBuffer::Borrowed(&task_type_hash[..]), + WriteBuffer::Borrowed(key), + )?; + } } } Ok(SnapshotMeta { data_items, meta_items, task_cache_items, - // The on-disk byte totals aren't known until the batch is committed below; - // they're filled in from `CommitStats` after `batch.commit()`. + // The on-disk byte totals aren't known until the batch is committed + // below; they're filled in from `CommitStats` after `batch.commit()`. bytes_written: 0, bytes_deleted: 0, max_next_task_id: max_new_task_id, @@ -294,13 +313,13 @@ impl TurboBackingStorage { .reduce(|t1, t2| t1.merge(t2)) .unwrap_or_default(); - let span = tracing::trace_span!("flush task data").entered(); + let span = tracing::trace_span!("flush task data"); parallel::try_for_each( &[KeySpace::TaskMeta, KeySpace::TaskData, KeySpace::TaskCache], |&key_space| { let _span = span.clone().entered(); - // Safety: `map_collect_owned` has returned, so no concurrent `put` or - // `delete` on these key spaces are in-flight. + // Safety: `map_collect_owned` has returned, so no concurrent `put` or `delete` + // on these key spaces are in-flight. unsafe { batch.flush(key_space) } }, )?; @@ -350,12 +369,16 @@ impl TurboBackingStorage { Ok(task_ids) } + /// Reads the stored `category` for `task_id`. + /// + /// `None` means the database had no key for it. That is distinct from `Some` of an empty + /// [`TaskStorage`] (a key that decoded to nothing), which is what lets a `MustExist` open tell + /// "absent everywhere" from "present but empty". pub(crate) fn lookup_data( &self, task_id: TaskId, category: SpecificTaskDataCategory, - storage: &mut TaskStorage, - ) -> Result<()> { + ) -> Result> { let inner = &*self.inner; let Some(bytes) = inner .database @@ -364,12 +387,14 @@ impl TurboBackingStorage { format!("Looking up task storage for {task_id} from database failed") })? else { - return Ok(()); + return Ok(None); }; + let mut storage = TaskStorage::default(); let mut decoder = new_turbo_bincode_decoder(bytes.borrow()); storage .decode(category, &mut decoder) - .map_err(|e| anyhow::anyhow!("Failed to decode {category:?}: {e:?}")) + .with_context(|| format!("Failed to decode {category:?}"))?; + Ok(Some(storage)) } pub(crate) fn batch_lookup_data( @@ -484,6 +509,20 @@ mod tests { Ok(()) } + /// Reads the TaskIds stored under `hash` in `TaskCache`, sorted for stable comparison. + fn task_cache_ids(db: &TurboKeyValueDatabase, hash: u64) -> Result> { + let mut ids: Vec = db + .get_multiple(KeySpace::TaskCache, &hash.to_le_bytes())? + .iter() + .map(|bytes| { + let bytes: [u8; 4] = Borrow::<[u8]>::borrow(bytes).try_into().unwrap(); + TaskId::try_from(u32::from_le_bytes(bytes)).unwrap() + }) + .collect(); + ids.sort_by_key(|id| **id); + Ok(ids) + } + /// Tests that `get_multiple` correctly returns multiple TaskIds when the same hash key /// is used (simulating a hash collision scenario). /// @@ -511,26 +550,12 @@ mod tests { write_task_cache_entry(&db, collision_hash, task_id_3)?; // Now query using get_multiple - should return all three TaskIds - let results = db.get_multiple(KeySpace::TaskCache, &collision_hash.to_le_bytes())?; - assert_eq!( - results.len(), - 3, + task_cache_ids(&db, collision_hash)?, + vec![task_id_1, task_id_2, task_id_3], "Should return all 3 task IDs for the colliding hash" ); - // Convert results to TaskIds and verify all three are present - let mut found_ids: Vec = results - .iter() - .map(|bytes| { - let bytes: [u8; 4] = Borrow::<[u8]>::borrow(bytes).try_into().unwrap(); - TaskId::try_from(u32::from_le_bytes(bytes)).unwrap() - }) - .collect(); - found_ids.sort_by_key(|id| **id); - - assert_eq!(found_ids, vec![task_id_1, task_id_2, task_id_3]); - db.shutdown()?; Ok(()) } @@ -589,4 +614,78 @@ mod tests { Ok(()) } + + /// `save_snapshot` delete path: a `Delete` item must erase the task's `TaskMeta` and + /// `TaskData` (`SingleValue`) entries and remove *only* that id from its `TaskCache` + /// (`MultiValue`) bucket. + /// + /// The colliding survivor is never read or rewritten — the key-value tombstone names the + /// single id it deletes, so anything else in the bucket is untouched whether or not this + /// commit knows about it. + #[tokio::test(flavor = "multi_thread")] + async fn test_save_snapshot_delete_tombstones_task() -> Result<()> { + let tempdir = tempfile::tempdir()?; + let path = tempdir.path(); + + let collision_hash: u64 = 0xC0FFEE; + let deleted_id = TaskId::try_from(111u32).unwrap(); + let survivor_id = TaskId::try_from(222u32).unwrap(); + let deleted_key = (*deleted_id).to_le_bytes(); + + let db = TurboKeyValueDatabase::new(path.to_path_buf(), false, true, false)?; + + // Both ids collide in one TaskCache bucket, purely on disk; the deleted task also has + // meta and data entries. + write_task_cache_entry(&db, collision_hash, deleted_id)?; + write_task_cache_entry(&db, collision_hash, survivor_id)?; + let batch = db.write_batch()?; + batch.put( + KeySpace::TaskMeta, + WriteBuffer::Borrowed(&deleted_key), + WriteBuffer::Borrowed(b"meta-bytes"), + )?; + batch.put( + KeySpace::TaskData, + WriteBuffer::Borrowed(&deleted_key), + WriteBuffer::Borrowed(b"data-bytes"), + )?; + batch.commit()?; + + // Sanity: everything is present before the delete. + assert!(db.get(KeySpace::TaskMeta, &deleted_key)?.is_some()); + assert!(db.get(KeySpace::TaskData, &deleted_key)?.is_some()); + assert_eq!( + task_cache_ids(&db, collision_hash)?, + vec![deleted_id, survivor_id], + ); + + let storage = TurboBackingStorage::new_in_memory(db); + + // Snapshot with no task data, just the one deletion. + storage.save_snapshot( + Vec::new(), + vec![vec![SnapshotItem::Delete { + task_id: deleted_id, + task_type_hash: collision_hash.to_le_bytes(), + }]], + )?; + + let db = &storage.inner.database; + assert!( + db.get(KeySpace::TaskMeta, &deleted_key)?.is_none(), + "TaskMeta should be tombstoned" + ); + assert!( + db.get(KeySpace::TaskData, &deleted_key)?.is_none(), + "TaskData should be tombstoned" + ); + assert_eq!( + task_cache_ids(db, collision_hash)?, + vec![survivor_id], + "save_snapshot should delete only the named id from the bucket" + ); + + db.shutdown()?; + Ok(()) + } } diff --git a/turbopack/crates/turbo-tasks-backend/tests/eviction.rs b/turbopack/crates/turbo-tasks-backend/tests/eviction.rs index e7712ca1b1bc..5cb01a81b78b 100644 --- a/turbopack/crates/turbo-tasks-backend/tests/eviction.rs +++ b/turbopack/crates/turbo-tasks-backend/tests/eviction.rs @@ -13,21 +13,17 @@ use turbo_tasks::{ }; use turbo_tasks_backend::{BackendOptions, EvictionMode, GitVersionInfo, TurboTasksBackend}; -/// Creates a fresh per-call persistence directory rooted under -/// `CARGO_TARGET_TMPDIR/.cache/`, with the test `name` as a prefix so failed -/// runs are easy to find on disk. The unique suffix from `tempfile` lets -/// multiple processes (or repeated invocations of the same test) run in -/// parallel without trampling each other's database. +/// Creates a fresh per-call persistence directory in the OS temp dir, with the test `name` as a +/// prefix so a leaked directory from a failed run is identifiable. The unique suffix from +/// `tempfile` lets repeated or concurrent runs coexist without trampling each other's database. /// -/// The returned [`tempfile::TempDir`] cleans up its contents on drop, so -/// callers should keep it alive at least until the `TurboTasks` it backs has -/// finished shutting down (so the final snapshot can flush to disk). +/// The returned [`tempfile::TempDir`] cleans up its contents on drop, so callers should keep it +/// alive at least until the `TurboTasks` it backs has finished shutting down (so the final snapshot +/// can flush to disk). fn create_test_persistence_dir(name: &str) -> tempfile::TempDir { - let parent = std::path::PathBuf::from(format!("{}/.cache", env!("CARGO_TARGET_TMPDIR"))); - std::fs::create_dir_all(&parent).unwrap(); tempfile::Builder::new() .prefix(&format!("{name}-")) - .tempdir_in(&parent) + .tempdir() .unwrap() }