turbo_tasks_backend/
kv_backing_storage.rs

1use std::{
2    borrow::Borrow,
3    env,
4    path::PathBuf,
5    sync::{Arc, LazyLock, Mutex, PoisonError, Weak},
6};
7
8use anyhow::{Context, Result};
9use smallvec::SmallVec;
10use turbo_bincode::{new_turbo_bincode_decoder, turbo_bincode_decode, turbo_bincode_encode};
11use turbo_persistence::CommitStats;
12use turbo_tasks::{
13    DynTaskInputs, RawVc, TaskId,
14    macro_helpers::NativeFunction,
15    panic_hooks::{PanicHookGuard, register_panic_hook},
16    parallel,
17};
18
19use crate::{
20    GitVersionInfo,
21    backend::{AnyOperation, SpecificTaskDataCategory, storage_schema::TaskStorage},
22    backing_storage::{SnapshotItem, SnapshotMeta, compute_task_type_hash_from_components},
23    database::{
24        db_invalidation::{StartupCacheState, check_db_invalidation_and_cleanup, invalidate_db},
25        db_versioning::handle_db_versioning,
26        key_value_database::KeySpace,
27        turbo::{TurboKeyValueDatabase, TurboWriteBatch},
28        write_batch::WriteBuffer,
29    },
30    db_invalidation::invalidation_reasons,
31};
32
33const META_KEY_OPERATIONS: u32 = 0;
34const META_KEY_NEXT_FREE_TASK_ID: u32 = 1;
35
36struct IntKey([u8; 4]);
37
38impl IntKey {
39    fn new(value: u32) -> Self {
40        Self(value.to_le_bytes())
41    }
42}
43
44impl AsRef<[u8]> for IntKey {
45    fn as_ref(&self) -> &[u8] {
46        &self.0
47    }
48}
49
50fn as_u32(bytes: impl Borrow<[u8]>) -> Result<u32> {
51    let n = u32::from_le_bytes(bytes.borrow().try_into()?);
52    Ok(n)
53}
54
55// We want to invalidate the cache on panic for most users, but this is a band-aid to underlying
56// problems in turbo-tasks.
57//
58// If we invalidate the cache upon panic and it "fixes" the issue upon restart, users typically
59// won't report bugs to us, and we'll never find root-causes for these problems.
60//
61// These overrides let us avoid the cache invalidation / error suppression within Vercel so that we
62// feel these pain points and fix the root causes of bugs.
63fn should_invalidate_on_panic() -> bool {
64    fn env_is_falsy(key: &str) -> bool {
65        env::var_os(key)
66            .is_none_or(|value| ["".as_ref(), "0".as_ref(), "false".as_ref()].contains(&&*value))
67    }
68    static SHOULD_INVALIDATE: LazyLock<bool> = LazyLock::new(|| {
69        env_is_falsy("TURBO_ENGINE_SKIP_INVALIDATE_ON_PANIC") && env_is_falsy("__NEXT_TEST_MODE")
70    });
71    *SHOULD_INVALIDATE
72}
73
74struct TurboBackingStorageInner {
75    database: TurboKeyValueDatabase,
76    /// Used when calling [`TurboBackingStorage::invalidate`]. Can be `None` in the
77    /// memory-only/no-op storage case.
78    base_path: Option<PathBuf>,
79    /// Used to skip calling [`invalidate_db`] when the database has already been invalidated.
80    invalidated: Mutex<bool>,
81    /// We configure a panic hook to invalidate the cache. This guard cleans up our panic hook upon
82    /// drop.
83    _panic_hook_guard: Option<PanicHookGuard>,
84}
85
86/// The higher-level backing storage passed to [`TurboTasksBackend::new`], used by
87/// [`crate::turbo_backing_storage`] and [`crate::noop_backing_storage`].
88///
89/// Wraps a low-level [`TurboKeyValueDatabase`] and adapts it into the persistence operations the
90/// backend needs (snapshots, task-candidate lookups, etc.).
91///
92/// [`TurboTasksBackend::new`]: crate::TurboTasksBackend::new
93pub struct TurboBackingStorage {
94    // wrapped so that `register_panic_hook` can hold a weak reference to `inner`.
95    inner: Arc<TurboBackingStorageInner>,
96}
97
98impl TurboBackingStorage {
99    pub(crate) fn new_in_memory(database: TurboKeyValueDatabase) -> Self {
100        Self {
101            inner: Arc::new(TurboBackingStorageInner {
102                database,
103                base_path: None,
104                invalidated: Mutex::new(false),
105                _panic_hook_guard: None,
106            }),
107        }
108    }
109
110    /// Handles boilerplate logic for an on-disk persisted database with versioning.
111    ///
112    /// - Creates a directory per version, with a maximum number of old versions and performs
113    ///   automatic cleanup of old versions.
114    /// - Checks for a database invalidation marker file, and cleans up the database as needed.
115    /// - [Registers a dynamic panic hook][turbo_tasks::panic_hooks] to invalidate the database upon
116    ///   a panic. This invalidates the database using [`invalidation_reasons::PANIC`].
117    ///
118    /// Along with returning a [`TurboBackingStorage`], this returns a
119    /// [`StartupCacheState`], which can be used by the application for logging information to the
120    /// user or telemetry about the cache.
121    pub(crate) fn open_versioned_on_disk(
122        base_path: PathBuf,
123        version_info: &GitVersionInfo,
124        is_ci: bool,
125        database: impl FnOnce(PathBuf) -> Result<TurboKeyValueDatabase>,
126    ) -> Result<(Self, StartupCacheState)> {
127        let startup_cache_state = check_db_invalidation_and_cleanup(&base_path)
128            .context("Failed to check database invalidation and cleanup")?;
129        let versioned_path = handle_db_versioning(&base_path, version_info, is_ci)
130            .context("Failed to handle database versioning")?;
131        let database = (database)(versioned_path).context("Failed to open database")?;
132        let backing_storage = Self {
133            inner: Arc::new_cyclic(move |weak_inner: &Weak<TurboBackingStorageInner>| {
134                let panic_hook_guard = if should_invalidate_on_panic() {
135                    let weak_inner = weak_inner.clone();
136                    Some(register_panic_hook(Box::new(move |_| {
137                        let Some(inner) = weak_inner.upgrade() else {
138                            return;
139                        };
140                        // If a panic happened that must mean something deep inside of turbopack
141                        // or turbo-tasks failed, and it may be hard to recover. We don't want
142                        // the cache to stick around, as that may persist bugs. Make a
143                        // best-effort attempt to invalidate the database (ignoring failures).
144                        let _ = inner.invalidate(invalidation_reasons::PANIC);
145                    })))
146                } else {
147                    None
148                };
149                TurboBackingStorageInner {
150                    database,
151                    base_path: Some(base_path),
152                    invalidated: Mutex::new(false),
153                    _panic_hook_guard: panic_hook_guard,
154                }
155            }),
156        };
157        Ok((backing_storage, startup_cache_state))
158    }
159}
160
161impl TurboBackingStorageInner {
162    fn invalidate(&self, reason_code: &str) -> Result<()> {
163        // `base_path` is `None` for in-memory backing storage (see `noop_backing_storage`).
164        if let Some(base_path) = &self.base_path {
165            // Invalidation could happen frequently if there's a bunch of panics. We only need to
166            // invalidate once, so grab a lock.
167            let mut invalidated_guard = self
168                .invalidated
169                .lock()
170                .unwrap_or_else(PoisonError::into_inner);
171            if *invalidated_guard {
172                return Ok(());
173            }
174            // Invalidate first, as it's a very fast atomic operation. `prevent_writes` is allowed
175            // to be slower (e.g. wait for a lock) and is allowed to corrupt the database with
176            // partial writes.
177            invalidate_db(base_path, reason_code)?;
178            self.database.prevent_writes();
179            // Avoid redundant invalidations from future panics
180            *invalidated_guard = true;
181        }
182        Ok(())
183    }
184
185    /// Used to read the next free task ID from the database.
186    fn get_infra_u32(&self, key: u32) -> Result<Option<u32>> {
187        self.database
188            .get(KeySpace::Infra, IntKey::new(key).as_ref())?
189            .map(as_u32)
190            .transpose()
191    }
192}
193
194impl TurboBackingStorage {
195    /// Called when the database should be invalidated upon re-initialization.
196    ///
197    /// This typically means that we'll restart the process or `turbo-tasks` soon with a fresh
198    /// database. If this happens, there's no point in writing anything else to disk, or flushing
199    /// during [`TurboTasksBackend::stop`].
200    ///
201    /// [`TurboTasksBackend::stop`]: turbo_tasks::backend::Backend::stop
202    pub(crate) fn invalidate(&self, reason_code: &str) -> Result<()> {
203        self.inner.invalidate(reason_code)
204    }
205
206    pub(crate) fn next_free_task_id(&self) -> Result<TaskId> {
207        Ok(self
208            .inner
209            .get_infra_u32(META_KEY_NEXT_FREE_TASK_ID)
210            .context("Unable to read next free task id from database")?
211            .map_or(Ok(TaskId::MIN), TaskId::try_from)?)
212    }
213
214    pub(crate) fn uncompleted_operations(&self) -> Result<Vec<AnyOperation>> {
215        fn get(database: &TurboKeyValueDatabase) -> Result<Vec<AnyOperation>> {
216            let Some(operations) =
217                database.get(KeySpace::Infra, IntKey::new(META_KEY_OPERATIONS).as_ref())?
218            else {
219                return Ok(Vec::new());
220            };
221            let operations = turbo_bincode_decode(operations.borrow())?;
222            Ok(operations)
223        }
224        get(&self.inner.database).context("Unable to read uncompleted operations from database")
225    }
226
227    pub(crate) fn save_snapshot<I>(
228        &self,
229        operations: Vec<Arc<AnyOperation>>,
230        snapshots: Vec<I>,
231    ) -> Result<SnapshotMeta>
232    where
233        I: IntoIterator<Item = SnapshotItem> + Send + Sync,
234    {
235        let _span = tracing::info_span!("save snapshot", operations = operations.len()).entered();
236        let batch = self.inner.database.write_batch()?;
237
238        {
239            let span = tracing::trace_span!("update task data");
240            let mut snapshot_meta =
241                parallel::map_collect_owned::<_, _, Result<Vec<_>>>(snapshots, |shard: I| {
242                    let _span = span.clone().entered();
243                    let mut max_new_task_id = 0;
244                    let mut data_items = 0;
245                    let mut meta_items = 0;
246                    let mut task_cache_items = 0;
247                    for item in shard {
248                        match item {
249                            SnapshotItem::Put {
250                                task_id,
251                                meta,
252                                data,
253                                task_type_hash,
254                            } => {
255                                let key = IntKey::new(*task_id);
256                                let key = key.as_ref();
257                                if let Some(meta) = meta {
258                                    batch.put(
259                                        KeySpace::TaskMeta,
260                                        WriteBuffer::Borrowed(key),
261                                        WriteBuffer::SmallVec(meta),
262                                    )?;
263                                    meta_items += 1;
264                                }
265                                if let Some(data) = data {
266                                    batch.put(
267                                        KeySpace::TaskData,
268                                        WriteBuffer::Borrowed(key),
269                                        WriteBuffer::SmallVec(data),
270                                    )?;
271                                    data_items += 1;
272                                }
273                                // Register the task type only for new tasks.
274                                if let Some(task_type_hash) = task_type_hash {
275                                    batch.put(
276                                        KeySpace::TaskCache,
277                                        WriteBuffer::Borrowed(&task_type_hash),
278                                        WriteBuffer::Borrowed(key),
279                                    )?;
280                                    task_cache_items += 1;
281                                    max_new_task_id = max_new_task_id.max(*task_id);
282                                }
283                            }
284                            SnapshotItem::Delete {
285                                task_id,
286                                task_type_hash,
287                            } => {
288                                let key = IntKey::new(*task_id);
289                                let key = key.as_ref();
290                                batch.delete(KeySpace::TaskMeta, WriteBuffer::Borrowed(key))?;
291                                batch.delete(KeySpace::TaskData, WriteBuffer::Borrowed(key))?;
292                                // TaskCache is MultiValue, delete just this id from the bucket.
293                                batch.delete_value(
294                                    KeySpace::TaskCache,
295                                    WriteBuffer::Borrowed(&task_type_hash[..]),
296                                    WriteBuffer::Borrowed(key),
297                                )?;
298                            }
299                        }
300                    }
301                    Ok(SnapshotMeta {
302                        data_items,
303                        meta_items,
304                        task_cache_items,
305                        // The on-disk byte totals aren't known until the batch is committed
306                        // below; they're filled in from `CommitStats` after `batch.commit()`.
307                        bytes_written: 0,
308                        bytes_deleted: 0,
309                        max_next_task_id: max_new_task_id,
310                    })
311                })?
312                .into_iter()
313                .reduce(|t1, t2| t1.merge(t2))
314                .unwrap_or_default();
315
316            let span = tracing::trace_span!("flush task data");
317            parallel::try_for_each(
318                &[KeySpace::TaskMeta, KeySpace::TaskData, KeySpace::TaskCache],
319                |&key_space| {
320                    let _span = span.clone().entered();
321                    // Safety: `map_collect_owned` has returned, so no concurrent `put` or `delete`
322                    // on these key spaces are in-flight.
323                    unsafe { batch.flush(key_space) }
324                },
325            )?;
326
327            let mut next_task_id = get_next_free_task_id(&batch)?;
328            next_task_id = next_task_id.max(snapshot_meta.max_next_task_id + 1);
329
330            save_infra(&batch, next_task_id, operations)?;
331            {
332                let _span = tracing::trace_span!("commit").entered();
333                // Byte totals are the physical on-disk bytes (post-compression, including .sst /
334                // .blob / .meta files) produced and removed by the commit.
335                let stats = batch.commit().context("Unable to commit snapshot")?;
336                snapshot_meta.bytes_written = stats.bytes_written;
337                snapshot_meta.bytes_deleted = stats.bytes_deleted;
338            }
339            Ok(snapshot_meta)
340        }
341    }
342
343    pub(crate) fn lookup_task_candidates(
344        &self,
345        native_fn: &'static NativeFunction,
346        this: Option<RawVc>,
347        arg: &dyn DynTaskInputs,
348    ) -> Result<SmallVec<[TaskId; 1]>> {
349        let inner = &*self.inner;
350        if inner.database.is_empty() {
351            // Checking if the database is empty is a performance optimization
352            // to avoid computing the hash.
353            return Ok(SmallVec::new());
354        }
355        let hash = compute_task_type_hash_from_components(native_fn, this, arg);
356        let buffers = inner
357            .database
358            .get_multiple(KeySpace::TaskCache, &hash)
359            .with_context(|| {
360                format!("Looking up task id for {native_fn:?}(this={this:?}) from database failed")
361            })?;
362
363        let mut task_ids = SmallVec::with_capacity(buffers.len());
364        for bytes in buffers {
365            let bytes = Borrow::<[u8]>::borrow(&bytes).try_into()?;
366            let id = TaskId::try_from(u32::from_le_bytes(bytes)).unwrap();
367            task_ids.push(id);
368        }
369        Ok(task_ids)
370    }
371
372    /// Reads the stored `category` for `task_id`.
373    ///
374    /// `None` means the database had no key for it. That is distinct from `Some` of an empty
375    /// [`TaskStorage`] (a key that decoded to nothing), which is what lets a `MustExist` open tell
376    /// "absent everywhere" from "present but empty".
377    pub(crate) fn lookup_data(
378        &self,
379        task_id: TaskId,
380        category: SpecificTaskDataCategory,
381    ) -> Result<Option<TaskStorage>> {
382        let inner = &*self.inner;
383        let Some(bytes) = inner
384            .database
385            .get(category.key_space(), IntKey::new(*task_id).as_ref())
386            .with_context(|| {
387                format!("Looking up task storage for {task_id} from database failed")
388            })?
389        else {
390            return Ok(None);
391        };
392        let mut storage = TaskStorage::default();
393        let mut decoder = new_turbo_bincode_decoder(bytes.borrow());
394        storage
395            .decode(category, &mut decoder)
396            .with_context(|| format!("Failed to decode {category:?}"))?;
397        Ok(Some(storage))
398    }
399
400    pub(crate) fn batch_lookup_data(
401        &self,
402        task_ids: &[TaskId],
403        category: SpecificTaskDataCategory,
404    ) -> Result<Vec<TaskStorage>> {
405        let inner = &*self.inner;
406        let int_keys: Vec<_> = task_ids.iter().map(|&id| IntKey::new(*id)).collect();
407        let keys = int_keys.iter().map(|k| k.as_ref()).collect::<Vec<_>>();
408        let bytes = inner
409            .database
410            .batch_get(category.key_space(), &keys)
411            .with_context(|| {
412                format!(
413                    "Looking up typed data for {} tasks from database failed",
414                    task_ids.len()
415                )
416            })?;
417        bytes
418            .into_iter()
419            .map(|opt_bytes| {
420                let mut storage = TaskStorage::new();
421                if let Some(bytes) = opt_bytes {
422                    let mut decoder = new_turbo_bincode_decoder(bytes.borrow());
423                    storage
424                        .decode(category, &mut decoder)
425                        .map_err(|e| anyhow::anyhow!("Failed to decode {category:?}: {e:?}"))?;
426                }
427                Ok(storage)
428            })
429            .collect::<Result<Vec<_>>>()
430    }
431
432    pub(crate) fn compact(&self) -> Result<Option<CommitStats>> {
433        self.inner.database.compact()
434    }
435
436    pub(crate) fn shutdown(&self) -> Result<()> {
437        self.inner.database.shutdown()
438    }
439
440    pub(crate) fn has_unrecoverable_write_error(&self) -> bool {
441        self.inner.database.has_unrecoverable_write_error()
442    }
443}
444
445fn get_next_free_task_id(batch: &TurboWriteBatch<'_>) -> Result<u32, anyhow::Error> {
446    Ok(
447        match batch.get(
448            KeySpace::Infra,
449            IntKey::new(META_KEY_NEXT_FREE_TASK_ID).as_ref(),
450        )? {
451            Some(bytes) => u32::from_le_bytes(Borrow::<[u8]>::borrow(&bytes).try_into()?),
452            None => 1,
453        },
454    )
455}
456
457fn save_infra(
458    batch: &TurboWriteBatch<'_>,
459    next_task_id: u32,
460    operations: Vec<Arc<AnyOperation>>,
461) -> Result<(), anyhow::Error> {
462    batch
463        .put(
464            KeySpace::Infra,
465            WriteBuffer::Borrowed(IntKey::new(META_KEY_NEXT_FREE_TASK_ID).as_ref()),
466            WriteBuffer::Borrowed(&next_task_id.to_le_bytes()),
467        )
468        .context("Unable to write next free task id")?;
469    {
470        let _span =
471            tracing::trace_span!("update operations", operations = operations.len()).entered();
472        let operations =
473            turbo_bincode_encode(&operations).context("Unable to serialize operations")?;
474        batch
475            .put(
476                KeySpace::Infra,
477                WriteBuffer::Borrowed(IntKey::new(META_KEY_OPERATIONS).as_ref()),
478                WriteBuffer::SmallVec(operations),
479            )
480            .context("Unable to write operations")?;
481    }
482    // Safety: save_infra is called after all concurrent writes to Infra are done.
483    unsafe { batch.flush(KeySpace::Infra)? };
484    Ok(())
485}
486
487#[cfg(test)]
488mod tests {
489    use std::borrow::Borrow;
490
491    use turbo_tasks::TaskId;
492
493    use super::*;
494    use crate::{
495        BackingStorageOptions,
496        database::{turbo::TurboKeyValueDatabase, write_batch::WriteBuffer},
497    };
498
499    /// Options used by these tests. `is_short_session` disables background compaction, which
500    /// requires a turbo-tasks context that these tests don't set up.
501    const TEST_STORAGE_OPTIONS: BackingStorageOptions = BackingStorageOptions {
502        is_ci: false,
503        is_short_session: true,
504        skip_compaction: false,
505    };
506
507    /// Helper to write to the database using the concurrent batch API.
508    fn write_task_cache_entry(
509        db: &TurboKeyValueDatabase,
510        hash: u64,
511        task_id: TaskId,
512    ) -> Result<()> {
513        let batch = db.write_batch()?;
514        batch.put(
515            KeySpace::TaskCache,
516            WriteBuffer::Borrowed(&hash.to_le_bytes()),
517            WriteBuffer::Borrowed(&(*task_id).to_le_bytes()),
518        )?;
519        batch.commit()?;
520        Ok(())
521    }
522
523    /// Reads the TaskIds stored under `hash` in `TaskCache`, sorted for stable comparison.
524    fn task_cache_ids(db: &TurboKeyValueDatabase, hash: u64) -> Result<Vec<TaskId>> {
525        let mut ids: Vec<TaskId> = db
526            .get_multiple(KeySpace::TaskCache, &hash.to_le_bytes())?
527            .iter()
528            .map(|bytes| {
529                let bytes: [u8; 4] = Borrow::<[u8]>::borrow(bytes).try_into().unwrap();
530                TaskId::try_from(u32::from_le_bytes(bytes)).unwrap()
531            })
532            .collect();
533        ids.sort_by_key(|id| **id);
534        Ok(ids)
535    }
536
537    /// Tests that `get_multiple` correctly returns multiple TaskIds when the same hash key
538    /// is used (simulating a hash collision scenario).
539    ///
540    /// This is a lower-level test that verifies the database layer correctly handles
541    /// the case where multiple task IDs are stored under the same hash key.
542    #[tokio::test(flavor = "multi_thread")]
543    async fn test_hash_collision_returns_multiple_candidates() -> Result<()> {
544        let tempdir = tempfile::tempdir()?;
545        let path = tempdir.path();
546
547        let db = TurboKeyValueDatabase::new(path.to_path_buf(), TEST_STORAGE_OPTIONS)?;
548
549        // Simulate a hash collision by writing multiple TaskIds with the same hash key
550        let collision_hash: u64 = 0xDEADBEEF;
551        let task_id_1 = TaskId::try_from(100u32).unwrap();
552        let task_id_2 = TaskId::try_from(200u32).unwrap();
553        let task_id_3 = TaskId::try_from(300u32).unwrap();
554
555        // Write three task IDs under the same hash key (simulating collision)
556        // Each write creates a new SST file, so all three will be returned by get_multiple
557        write_task_cache_entry(&db, collision_hash, task_id_1)?;
558        write_task_cache_entry(&db, collision_hash, task_id_2)?;
559        write_task_cache_entry(&db, collision_hash, task_id_3)?;
560
561        // Now query using get_multiple - should return all three TaskIds
562        assert_eq!(
563            task_cache_ids(&db, collision_hash)?,
564            vec![task_id_1, task_id_2, task_id_3],
565            "Should return all 3 task IDs for the colliding hash"
566        );
567
568        db.shutdown()?;
569        Ok(())
570    }
571
572    /// Tests that multiple distinct keys written in a single batch with flush can be read back.
573    /// This mirrors the actual save_snapshot pattern: write many TaskCache entries, flush, commit.
574    #[tokio::test(flavor = "multi_thread")]
575    async fn test_batch_write_with_flush_and_reopen() -> Result<()> {
576        let tempdir = tempfile::tempdir()?;
577        let path = tempdir.path();
578
579        let n = 100_000;
580        let hashes: Vec<u64> = (0..n).map(|i| 0x1000 + i as u64).collect();
581        let task_ids: Vec<TaskId> = (1..=n as u32)
582            .map(|i| TaskId::try_from(i).unwrap())
583            .collect();
584
585        // Write all entries in a single batch with flush (like save_snapshot does)
586        {
587            let db = TurboKeyValueDatabase::new(path.to_path_buf(), TEST_STORAGE_OPTIONS)?;
588            let batch = db.write_batch()?;
589
590            for (hash, task_id) in hashes.iter().zip(task_ids.iter()) {
591                batch.put(
592                    KeySpace::TaskCache,
593                    WriteBuffer::Borrowed(&hash.to_le_bytes()),
594                    WriteBuffer::Borrowed(&(**task_id).to_le_bytes()),
595                )?;
596            }
597            // Flush TaskCache (like the new code does)
598            unsafe { batch.flush(KeySpace::TaskCache) }?;
599            batch.commit()?;
600
601            db.shutdown()?;
602        }
603
604        // Reopen and verify all entries are readable
605        {
606            let db = TurboKeyValueDatabase::new(path.to_path_buf(), TEST_STORAGE_OPTIONS)?;
607            let mut found = 0;
608            let mut missing = 0;
609            for (hash, expected_id) in hashes.iter().zip(task_ids.iter()) {
610                let results = db.get_multiple(KeySpace::TaskCache, &hash.to_le_bytes())?;
611                if results.is_empty() {
612                    missing += 1;
613                } else {
614                    found += 1;
615                    let bytes: [u8; 4] = Borrow::<[u8]>::borrow(&results[0]).try_into().unwrap();
616                    let id = TaskId::try_from(u32::from_le_bytes(bytes)).unwrap();
617                    assert_eq!(id, *expected_id, "Task ID mismatch for hash {hash:#x}");
618                }
619            }
620            assert_eq!(missing, 0, "Found {found}/{n} entries, missing {missing}");
621            db.shutdown()?;
622        }
623
624        Ok(())
625    }
626
627    /// `save_snapshot` delete path: a `Delete` item must erase the task's `TaskMeta` and
628    /// `TaskData` (`SingleValue`) entries and remove *only* that id from its `TaskCache`
629    /// (`MultiValue`) bucket.
630    ///
631    /// The colliding survivor is never read or rewritten — the key-value tombstone names the
632    /// single id it deletes, so anything else in the bucket is untouched whether or not this
633    /// commit knows about it.
634    #[tokio::test(flavor = "multi_thread")]
635    async fn test_save_snapshot_delete_tombstones_task() -> Result<()> {
636        let tempdir = tempfile::tempdir()?;
637        let path = tempdir.path();
638
639        let collision_hash: u64 = 0xC0FFEE;
640        let deleted_id = TaskId::try_from(111u32).unwrap();
641        let survivor_id = TaskId::try_from(222u32).unwrap();
642        let deleted_key = (*deleted_id).to_le_bytes();
643
644        let db = TurboKeyValueDatabase::new(
645            path.to_path_buf(),
646            BackingStorageOptions {
647                is_ci: false,
648                is_short_session: true,
649                skip_compaction: false,
650            },
651        )?;
652
653        // Both ids collide in one TaskCache bucket, purely on disk; the deleted task also has
654        // meta and data entries.
655        write_task_cache_entry(&db, collision_hash, deleted_id)?;
656        write_task_cache_entry(&db, collision_hash, survivor_id)?;
657        let batch = db.write_batch()?;
658        batch.put(
659            KeySpace::TaskMeta,
660            WriteBuffer::Borrowed(&deleted_key),
661            WriteBuffer::Borrowed(b"meta-bytes"),
662        )?;
663        batch.put(
664            KeySpace::TaskData,
665            WriteBuffer::Borrowed(&deleted_key),
666            WriteBuffer::Borrowed(b"data-bytes"),
667        )?;
668        batch.commit()?;
669
670        // Sanity: everything is present before the delete.
671        assert!(db.get(KeySpace::TaskMeta, &deleted_key)?.is_some());
672        assert!(db.get(KeySpace::TaskData, &deleted_key)?.is_some());
673        assert_eq!(
674            task_cache_ids(&db, collision_hash)?,
675            vec![deleted_id, survivor_id],
676        );
677
678        let storage = TurboBackingStorage::new_in_memory(db);
679
680        // Snapshot with no task data, just the one deletion.
681        storage.save_snapshot(
682            Vec::new(),
683            vec![vec![SnapshotItem::Delete {
684                task_id: deleted_id,
685                task_type_hash: collision_hash.to_le_bytes(),
686            }]],
687        )?;
688
689        let db = &storage.inner.database;
690        assert!(
691            db.get(KeySpace::TaskMeta, &deleted_key)?.is_none(),
692            "TaskMeta should be tombstoned"
693        );
694        assert!(
695            db.get(KeySpace::TaskData, &deleted_key)?.is_none(),
696            "TaskData should be tombstoned"
697        );
698        assert_eq!(
699            task_cache_ids(db, collision_hash)?,
700            vec![survivor_id],
701            "save_snapshot should delete only the named id from the bucket"
702        );
703
704        db.shutdown()?;
705        Ok(())
706    }
707}
turbo_tasks_backend/kv_backing_storage.rs

turbo_tasks_backend/
kv_backing_storage.rs
