Turbopack: skip db lookups on the initial build (#82405)

### What?

In certain cases (e.g. on initial build) the backend knows that the backing storage has no more info than the memory cache. In these cases it can skip looking up every task on the backing storage.

Author: sokra

turbopack/crates/turbo-tasks-backend/src/backend/mod.rs

@@ -176,6 +176,9 @@ struct TurboTasksBackendInner<B: BackingStorage> {
 
     storage: Storage,
 
+    /// When true, the backing_storage has data that is not in the local storage.
+    local_is_partial: AtomicBool,
+
     /// Number of executing operations + Highest bit is set when snapshot is
     /// requested. When that bit is set, operations should pause until the
     /// snapshot is completed. When the bit is set and in progress counter
@@ -231,16 +234,17 @@ impl<B: BackingStorage> TurboTasksBackendInner<B> {
             options.active_tracking = false;
         }
         let small_preallocation = options.small_preallocation;
+        let next_task_id = backing_storage
+            .next_free_task_id()
+            .expect("Failed to get task id");
         Self {
             options,
             start_time: Instant::now(),
             session_id: backing_storage
                 .next_session_id()
                 .expect("Failed get session id"),
             persisted_task_id_factory: IdFactoryWithReuse::new(
-                backing_storage
-                    .next_free_task_id()
-                    .expect("Failed to get task id"),
+                next_task_id,
                 TaskId::try_from(TRANSIENT_TASK_BIT - 1).unwrap(),
             ),
             transient_task_id_factory: IdFactoryWithReuse::new(
@@ -250,6 +254,7 @@ impl<B: BackingStorage> TurboTasksBackendInner<B> {
             persisted_task_cache_log: need_log.then(|| Sharded::new(shard_amount)),
             task_cache: BiMap::new(),
             transient_tasks: FxDashMap::default(),
+            local_is_partial: AtomicBool::new(next_task_id != TaskId::MIN),
             storage: Storage::new(small_preallocation),
             in_progress_operations: AtomicUsize::new(0),
             snapshot_request: Mutex::new(SnapshotRequest::new()),
@@ -910,6 +915,7 @@ impl<B: BackingStorage> TurboTasksBackendInner<B> {
             return Some(task_type);
         }
         if self.should_restore()
+            && self.local_is_partial.load(Ordering::Acquire)
             && !task_id.is_transient()
             && let Some(task_type) = unsafe {
                 self.backing_storage
@@ -1271,16 +1277,21 @@ impl<B: BackingStorage> TurboTasksBackendInner<B> {
             return task_id;
         }
 
-        let tx = self
-            .should_restore()
+        let check_backing_storage =
+            self.should_restore() && self.local_is_partial.load(Ordering::Acquire);
+        let tx = check_backing_storage
             .then(|| self.backing_storage.start_read_transaction())
             .flatten();
         let task_id = {
             // Safety: `tx` is a valid transaction from `self.backend.backing_storage`.
             if let Some(task_id) = unsafe {
-                self.backing_storage
-                    .forward_lookup_task_cache(tx.as_ref(), &task_type)
-                    .expect("Failed to lookup task id")
+                check_backing_storage
+                    .then(|| {
+                        self.backing_storage
+                            .forward_lookup_task_cache(tx.as_ref(), &task_type)
+                            .expect("Failed to lookup task id")
+                    })
+                    .flatten()
             } {
                 self.track_cache_hit(&task_type);
                 let _ = self.task_cache.try_insert(Arc::new(task_type), task_id);

turbopack/crates/turbo-tasks-backend/src/backend/operation/mod.rs

@@ -11,6 +11,7 @@ mod update_output;
 use std::{
     fmt::{Debug, Formatter},
     mem::transmute,
+    sync::atomic::Ordering,
 };
 
 use serde::{Deserialize, Serialize};
@@ -113,6 +114,12 @@ where
         category: TaskDataCategory,
     ) -> Vec<CachedDataItem> {
         if matches!(self.transaction, TransactionState::None) {
+            let check_backing_storage = self.backend.should_restore()
+                && self.backend.local_is_partial.load(Ordering::Acquire);
+            if !check_backing_storage {
+                // If we don't need to restore, we can just return an empty vector
+                return Vec::new();
+            }
             let tx = self.backend.backing_storage.start_read_transaction();
             let tx = tx.map(|tx| {
                 // Safety: self is actually valid for 'a, so it's safe to transmute 'l to 'a

turbopack/crates/turbo-tasks-backend/src/kv_backing_storage.rs

@@ -262,12 +262,11 @@ impl<T: KeyValueDatabase + Send + Sync + 'static> BackingStorageSealed
     type ReadTransaction<'l> = T::ReadTransaction<'l>;
 
     fn next_free_task_id(&self) -> Result<TaskId> {
-        Ok(TaskId::try_from(
-            self.inner
-                .get_infra_u32(META_KEY_NEXT_FREE_TASK_ID)
-                .context("Unable to read next free task id from database")?
-                .unwrap_or(1),
-        )?)
+        Ok(self
+            .inner
+            .get_infra_u32(META_KEY_NEXT_FREE_TASK_ID)
+            .context("Unable to read next free task id from database")?
+            .map_or(Ok(TaskId::MIN), TaskId::try_from)?)
     }
 
     fn next_session_id(&self) -> Result<SessionId> {