Optimize how we track data for persistence (#89370)

## Summary

Reworks how the turbo-tasks backend tracks modified tasks for persistence snapshots, reducing overhead and simplifying the snapshot lifecycle.

**Key changes:**

- **Replace `modified` DashMap with per-shard atomic counters + inline flags.** Instead of maintaining a separate `FxDashMap<TaskId, ModifiedState>` that mirrors every modification, track modifications via flags already on `TaskStorage` and use per-shard `AtomicU64` counters to skip unmodified shards during snapshot iteration. This eliminates a major source of  memory overhead.
    - The downside here is needing to scan the map.  per shard counters enable early exits but we will still need to scan entire shards.  For a large site that means scanning thousands of

- **Merge task cache writes into the snapshot pipeline.** New tasks now carry a `new_task` flag and their type hash in `SnapshotItem`, so task cache entries are written in the same batch as task data/meta — removing the separate `persisted_task_cache_log` (`Sharded<ChunkedVec<...>>`) and its associated locking.

- **Remove `local_is_partial` optimization.** The backing storage layer already short-circuits on empty databases, and new tasks eagerly set `restored` flags at allocation time, making this redundant.

- **Simplify `end_snapshot`.** Instead of a multi-pass retain/iterate/update cycle over the `modified` map, `end_snapshot` now just drains the small `snapshots` map (only tasks concurrently accessed during snapshot mode) and promotes their `modified_during_snapshot` flags.

- **Delete unused utilities.** Removes `Sharded`, `ChunkedVec` (from backing_storage), and `swap_retain` import now that they're no longer needed.

**Other cleanups:**
- `initialize_new_task` sets restored + new_task flags at allocation time for both persistent and transient tasks
- Fuzz test updated to use `active_tracking: true` and `StorageMode::ReadWrite`
- New KV storage test for batch write+flush+reopen pattern
- Minor fix: `SmallVec::into_boxed_slice()` instead of `into_vec().into_boxed_slice()`


## Build Benchmark Results

  Measured over 9 runs (1 warm-up discarded), macOS, `TURBOPACK_PERSISTENT_CACHE=1`.

  ### Cold build (`rm -rf .next/`)

  | | Time (avg) | Time (stddev) | MaxRSS (avg) |
  |---|---|---|---|
  | HEAD | 75.48s | 0.72s | 24,231 MiB |
  | This PR | 75.20s | 1.98s | 23,829 MiB |
  | **Delta** | −0.28s | — | **−402 MiB (−1.7%)** |

  ### Warm build (single file edit)

  | | Time (avg) | MaxRSS (avg) |
  |---|---|---|
  | HEAD | 23.79s | 8,764 MiB |
  | This PR | 23.67s | 8,766 MiB |
  | **Delta** | −0.12s | flat |

  Cold build time difference is within noise (< 1 stddev). The meaningful improvement is a **~400 MiB reduction in peak memory on cold builds**, consistent with the fixed overhead this PR targets. Warm builds are unaffected as expected.

Author: lukesandberg

test/e2e/filesystem-cache/next.config.js

@@ -17,14 +17,10 @@ const nextConfig = {
       },
     },
   },
-  experimental: enableCaching
-    ? {
-        turbopackFileSystemCacheForBuild: true,
-      }
-    : {
-        turbopackFileSystemCacheForDev: false,
-        turbopackFileSystemCacheForBuild: false,
-      },
+  experimental: {
+    turbopackFileSystemCacheForBuild: enableCaching,
+    turbopackFileSystemCacheForDev: enableCaching,
+  },
   env: {
     NEXT_PUBLIC_CONFIG_ENV: 'hello world',
   },

test/integration/config-output-export/test/index.test.ts

@@ -23,11 +23,9 @@ const runDev = async (config: any, shouldWaitForReady = true) => {
   const port = await findPort()
   const obj = { port, stdout: '', stderr: '' }
   app = await launchApp(appDir, port, {
-    stdout: false,
     onStdout(msg: string) {
       obj.stdout += msg || ''
     },
-    stderr: false,
     onStderr(msg: string) {
       obj.stderr += msg || ''
     },

turbopack/crates/turbo-persistence/src/value_buf.rs

@@ -13,7 +13,7 @@ impl ValueBuffer<'_> {
         match self {
             ValueBuffer::Borrowed(b) => b.into(),
             ValueBuffer::Vec(v) => v.into_boxed_slice(),
-            ValueBuffer::SmallVec(sv) => sv.into_vec().into_boxed_slice(),
+            ValueBuffer::SmallVec(sv) => sv.into_boxed_slice(),
         }
     }
 }

turbopack/crates/turbo-tasks-backend/fuzz/Cargo.toml

@@ -16,7 +16,7 @@ libfuzzer-sys = "0.4.9"
 once_cell = { workspace = true }
 serde = { workspace = true }
 tokio = { workspace = true, features = ["full"] }
-turbo-tasks = { workspace = true }
+turbo-tasks = { workspace = true, features = ["non_operation_vc_strongly_consistent"] }
 turbo-tasks-backend = { workspace = true }
 turbo-tasks-malloc = { workspace = true }
 

turbopack/crates/turbo-tasks-backend/fuzz/src/graph.rs

@@ -133,16 +133,17 @@ pub fn run(data: Vec<TaskSpec>) {
 struct Iteration(State<usize>);
 
 fn actual_operation(spec: Arc<Vec<TaskSpec>>, iterations: usize) {
-    let tt = TurboTasks::new(turbo_tasks_backend::TurboTasksBackend::new(
-        turbo_tasks_backend::BackendOptions {
-            storage_mode: None,
-            small_preallocation: true,
-            ..Default::default()
-        },
-        turbo_tasks_backend::noop_backing_storage(),
-    ));
     RUNTIME
         .block_on(async {
+            let tt = TurboTasks::new(turbo_tasks_backend::TurboTasksBackend::new(
+                turbo_tasks_backend::BackendOptions {
+                    storage_mode: Some(turbo_tasks_backend::StorageMode::ReadWrite),
+                    small_preallocation: false,
+                    active_tracking: true,
+                    ..Default::default()
+                },
+                turbo_tasks_backend::noop_backing_storage(),
+            ));
             for i in 0..iterations {
                 let spec = spec.clone();
                 tt.run(async move {

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

@@ -63,21 +63,18 @@ use crate::{
         storage::Storage,
         storage_schema::{TaskStorage, TaskStorageAccessors},
     },
-    backing_storage::{BackingStorage, SnapshotItem},
+    backing_storage::{BackingStorage, SnapshotItem, compute_task_type_hash},
     data::{
         ActivenessState, CellRef, CollectibleRef, CollectiblesRef, Dirtyness, InProgressCellState,
         InProgressState, InProgressStateInner, OutputValue, TransientTask,
     },
     error::TaskError,
     utils::{
         arc_or_owned::ArcOrOwned,
-        chunked_vec::ChunkedVec,
         dash_map_drop_contents::drop_contents,
         dash_map_raw_entry::{RawEntry, raw_entry},
         ptr_eq_arc::PtrEqArc,
         shard_amount::compute_shard_amount,
-        sharded::Sharded,
-        swap_retain,
     },
 };
 
@@ -167,8 +164,6 @@ pub enum TurboTasksBackendJob {
 
 pub struct TurboTasksBackend<B: BackingStorage>(Arc<TurboTasksBackendInner<B>>);
 
-type TaskCacheLog = Sharded<ChunkedVec<(Arc<CachedTaskType>, TaskId)>>;
-
 struct TurboTasksBackendInner<B: BackingStorage> {
     options: BackendOptions,
 
@@ -177,15 +172,10 @@ struct TurboTasksBackendInner<B: BackingStorage> {
     persisted_task_id_factory: IdFactoryWithReuse<TaskId>,
     transient_task_id_factory: IdFactoryWithReuse<TaskId>,
 
-    persisted_task_cache_log: Option<TaskCacheLog>,
     task_cache: FxDashMap<Arc<CachedTaskType>, TaskId>,
 
     storage: Storage,
 
-    /// When true, the backing_storage has data that is not in the local storage.
-    /// This is determined once at startup and never changes.
-    local_is_partial: bool,
-
     /// 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
@@ -235,10 +225,6 @@ impl<B: BackingStorage> TurboTasksBackend<B> {
 impl<B: BackingStorage> TurboTasksBackendInner<B> {
     pub fn new(mut options: BackendOptions, backing_storage: B) -> Self {
         let shard_amount = compute_shard_amount(options.num_workers, options.small_preallocation);
-        let need_log = matches!(
-            options.storage_mode,
-            Some(StorageMode::ReadWrite) | Some(StorageMode::ReadWriteOnShutdown)
-        );
         if !options.dependency_tracking {
             options.active_tracking = false;
         }
@@ -257,9 +243,7 @@ impl<B: BackingStorage> TurboTasksBackendInner<B> {
                 TaskId::try_from(TRANSIENT_TASK_BIT).unwrap(),
                 TaskId::MAX,
             ),
-            persisted_task_cache_log: need_log.then(|| Sharded::new(shard_amount)),
             task_cache: FxDashMap::default(),
-            local_is_partial: next_task_id != TaskId::MIN,
             storage: Storage::new(shard_amount, small_preallocation),
             in_progress_operations: AtomicUsize::new(0),
             snapshot_request: Mutex::new(SnapshotRequest::new()),
@@ -1036,12 +1020,9 @@ impl<B: BackingStorage> TurboTasksBackendInner<B> {
                 .map(|op| op.arc().clone())
                 .collect::<Vec<_>>();
         }
-        self.storage.start_snapshot();
-        let mut persisted_task_cache_log = self
-            .persisted_task_cache_log
-            .as_ref()
-            .map(|l| l.take(|i| i))
-            .unwrap_or_default();
+        // Enter snapshot mode, which atomically reads and resets the modified count.
+        // Checking after start_snapshot ensures no concurrent increments can race.
+        let (snapshot_guard, has_modifications) = self.storage.start_snapshot();
         let mut snapshot_request = self.snapshot_request.lock();
         snapshot_request.snapshot_requested = false;
         self.in_progress_operations
@@ -1050,6 +1031,13 @@ impl<B: BackingStorage> TurboTasksBackendInner<B> {
         let snapshot_time = Instant::now();
         drop(snapshot_request);
 
+        if !has_modifications {
+            // No tasks modified since the last snapshot — drop the guard (which
+            // calls end_snapshot) and skip the expensive O(N) scan.
+            drop(snapshot_guard);
+            return Some((start, false));
+        }
+
         #[cfg(feature = "print_cache_item_size")]
         #[derive(Default)]
         struct TaskCacheStats {
@@ -1278,35 +1266,49 @@ impl<B: BackingStorage> TurboTasksBackendInner<B> {
             } else {
                 None
             };
+            let task_type_hash = if inner.flags.new_task() {
+                let task_type = inner.get_persistent_task_type().expect(
+                    "It is not possible for a new_task to not have a persistent_task_type.  Task \
+                     creation for persistent tasks uses a single ExecutionContextImpl for \
+                     creating the task (which sets new_task) and connect_child (which sets \
+                     persistent_task_type) and take_snapshot waits for all operations to complete \
+                     or suspend before we start snapshotting.  So task creation will always set \
+                     the task_type.",
+                );
+                Some(compute_task_type_hash(task_type))
+            } else {
+                None
+            };
 
             SnapshotItem {
                 task_id,
                 meta,
                 data,
+                task_type_hash,
             }
         };
 
         // take_snapshot already filters empty items and empty shards in parallel
-        let task_snapshots = self.storage.take_snapshot(&process);
-
-        swap_retain(&mut persisted_task_cache_log, |shard| !shard.is_empty());
+        let task_snapshots = self.storage.take_snapshot(snapshot_guard, &process);
 
         drop(snapshot_span);
         let snapshot_duration = start.elapsed();
         let task_count = task_snapshots.len();
 
-        if persisted_task_cache_log.is_empty() && task_snapshots.is_empty() {
+        if task_snapshots.is_empty() {
+            // This should be impossible — if we got here, modified_count was nonzero, and every
+            // modification that increments the count also failed during encoding.
+            std::hint::cold_path();
             return Some((snapshot_time, false));
         }
 
         let persist_start = Instant::now();
         let _span = tracing::info_span!(parent: parent_span, "persist", reason = reason).entered();
         {
-            if let Err(err) = self.backing_storage.save_snapshot(
-                suspended_operations,
-                persisted_task_cache_log,
-                task_snapshots,
-            ) {
+            if let Err(err) = self
+                .backing_storage
+                .save_snapshot(suspended_operations, task_snapshots)
+            {
                 eprintln!("Persisting failed: {err:?}");
                 return None;
             }
@@ -1551,7 +1553,7 @@ impl<B: BackingStorage> TurboTasksBackendInner<B> {
             let (task_id, task_type) = match raw_entry(&self.task_cache, &task_type) {
                 RawEntry::Occupied(e) => {
                     // Another thread beat us to creating this task - use their task_id.
-                    // They will handle logging to persisted_task_cache_log.
+                    // They will handle logging the new task as modified
                     let task_id = *e.get();
                     drop(e);
                     self.track_cache_hit(&task_type);
@@ -1562,12 +1564,12 @@ impl<B: BackingStorage> TurboTasksBackendInner<B> {
                     let task_type = Arc::new(task_type);
                     let task_id = self.persisted_task_id_factory.get();
                     e.insert(task_type.clone(), task_id);
+                    // Mark the task as new in storage.
+                    // Do this after e.insert so we aren't holding the task_cache lock
+                    self.storage.initialize_new_task(task_id);
                     // insert() consumes e, releasing the lock
                     self.track_cache_miss(&task_type);
                     is_new = true;
-                    if let Some(log) = &self.persisted_task_cache_log {
-                        log.lock(task_id).push((task_type.clone(), task_id));
-                    }
                     (task_id, ArcOrOwned::Arc(task_type))
                 }
             };
@@ -1637,6 +1639,7 @@ impl<B: BackingStorage> TurboTasksBackendInner<B> {
                 let task_type = Arc::new(task_type);
                 let task_id = self.transient_task_id_factory.get();
                 e.insert(task_type.clone(), task_id);
+                self.storage.initialize_new_task(task_id);
                 self.track_cache_miss(&task_type);
 
                 if is_root {

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

@@ -1896,7 +1896,7 @@ impl AggregationUpdateQueue {
     /// See detailed comments in that function, follow the STEP numbers.
     fn inner_of_uppers_has_new_follower<T: TaskIdWithOptionalCount + Clone, const N: usize>(
         &mut self,
-        ctx: &mut impl ExecuteContext,
+        ctx: &mut impl ExecuteContext<'_>,
         new_follower_id: TaskId,
         mut upper_ids: SmallVec<[T; N]>,
     ) {
@@ -2618,7 +2618,7 @@ impl AggregationUpdateQueue {
     /// Only used when activeness is tracked.
     fn increase_active_count(
         &mut self,
-        ctx: &mut impl ExecuteContext,
+        ctx: &mut impl ExecuteContext<'_>,
         task_id: TaskId,
         task_type: Option<Arc<CachedTaskType>>,
     ) {
@@ -2637,7 +2637,7 @@ impl AggregationUpdateQueue {
         if let Some(task_type) = task_type
             && !task.has_persistent_task_type()
         {
-            let _ = task.set_persistent_task_type(task_type);
+            task.set_persistent_task_type(task_type);
         }
         let state = task.get_activeness_mut_or_insert_with(|| ActivenessState::new(task_id));
         let is_new = state.is_empty();