Repository: vercel/next.js
Pull request: Move database compaction from write batch commit to backend idle loop

## What?

Moves TurboPersistence database compaction out of the write batch commit path and into the backend's idle loop, where it can run repeatedly (up to 10 times) while the system is idle.

## Why?

Previously, compaction was kicked off as a background task after each write batch commit, with the next write batch blocking until the previous compaction finished. This had several drawbacks:

- Compaction was tightly coupled to the write path, adding latency to write batches that had to wait for the previous compaction to complete before starting.
- Only a single compaction pass ran per write batch, which may not be sufficient to fully compact the database (e.g., after many small writes create numerous segments).
- The compaction used `spawn` and a `JoinHandle` stored in a `Mutex`, adding complexity to the write batch and shutdown logic.

By moving compaction to the idle loop, we:
- Decouple compaction from writes, so write batches are never blocked by ongoing compaction.
- Allow multiple compaction passes (up to 10) while idle, enabling the database to converge to a more compact state.
- Simplify the write batch code by removing the `compact_join_handle` machinery (`Mutex`, `JoinHandle`, `spawn`).
- Respect idle-end signals — compaction stops immediately when new work arrives.

## How?

1. **Added `compact()` method** to the `KeyValueDatabase` trait and `BackingStorageSealed` trait, with a default no-op implementation returning `Ok(false)`. Returns `Ok(true)` when compaction actually merged files, `Ok(false)` when there was nothing to compact.

2. **Implemented `compact()` for `TurboKeyValueDatabase`** — calls the existing `do_compact` helper synchronously, skipping compaction for short sessions or empty databases. Changed `do_compact` return type from `Result<()>` to `Result<bool>` to signal whether work was done.

3. **Wired `compact()` through `KvBackingStorage`** and the `Either`-based `BackingStorageSealed` impl.

4. **Added compaction loop in the backend idle handler** (`backend/mod.rs`) — after a successful snapshot commit, runs up to 10 compaction passes, checking for idle-end between each pass and stopping early if no more compaction is needed or idle ends.

5. **Removed from `TurboWriteBatch`**: the `compact_join_handle` field, the `Mutex<Option<JoinHandle>>`, the background `spawn` call in `commit()`, and the blocking join in `write_batch()` and `shutdown()`. The `shutdown()` path retains its own compaction call for final compaction on exit.

Changed files:
filename: turbopack/crates/turbo-tasks-backend/src/backend/mod.rs
patch: @@ -2797,6 +2797,32 @@ impl<B: BackingStorage> TurboTasksBackendInner<B> {
                         let snapshot = this.snapshot_and_persist(None, reason, turbo_tasks);
                         if let Some((snapshot_start, new_data)) = snapshot {
                             last_snapshot = snapshot_start;
+
+                            // Compact while idle (up to limit), regardless of
+                            // whether the snapshot had new data.
+                            const MAX_IDLE_COMPACTION_PASSES: usize = 10;
+                            for _ in 0..MAX_IDLE_COMPACTION_PASSES {
+                                let idle_ended = tokio::select! {
+                                    biased;
+                                    _ = &mut idle_end_listener => {
+                                        idle_end_listener = self.idle_end_event.listen();
+                                        true
+                                    },
+                                    _ = std::future::ready(()) => false,
+                                };
+                                if idle_ended {
+                                    break;
+                                }
+                                match self.backing_storage.compact() {
+                                    Ok(true) => {}
+                                    Ok(false) => break,
+                                    Err(err) => {
+                                        eprintln!("Compaction failed: {err:?}");
+                                        break;
+                                    }
+                                }
+                            }
+
                             if !new_data {
                                 fresh_idle = false;
                                 continue;

filename: turbopack/crates/turbo-tasks-backend/src/backing_storage.rs
patch: @@ -87,6 +87,10 @@ pub trait BackingStorageSealed: 'static + Send + Sync {
         category: SpecificTaskDataCategory,
     ) -> Result<Vec<TaskStorage>>;
 
+    fn compact(&self) -> Result<bool> {
+        Ok(false)
+    }
+
     fn shutdown(&self) -> Result<()> {
         Ok(())
     }
@@ -152,6 +156,10 @@ where
         either::for_both!(self, this => this.batch_lookup_data(task_ids, category))
     }
 
+    fn compact(&self) -> Result<bool> {
+        either::for_both!(self, this => this.compact())
+    }
+
     fn shutdown(&self) -> Result<()> {
         either::for_both!(self, this => this.shutdown())
     }

filename: turbopack/crates/turbo-tasks-backend/src/database/key_value_database.rs
patch: @@ -82,6 +82,14 @@ pub trait KeyValueDatabase {
         // this is an optional performance hint to the database
     }
 
+    /// Triggers compaction of the database.
+    ///
+    /// Returns `Ok(true)` if compaction actually merged files, `Ok(false)` if there was nothing
+    /// to compact. The default implementation is a no-op.
+    fn compact(&self) -> Result<bool> {
+        Ok(false)
+    }
+
     fn shutdown(&self) -> Result<()> {
         Ok(())
     }

filename: turbopack/crates/turbo-tasks-backend/src/database/turbo/mod.rs
patch: @@ -7,15 +7,13 @@ use std::{
 };
 
 use anyhow::{Ok, Result};
-use parking_lot::Mutex;
 use smallvec::SmallVec;
 use turbo_persistence::{
     ArcBytes, CompactConfig, DbConfig, KeyBase, StoreKey, TurboPersistence, ValueBuffer,
 };
 use turbo_tasks::{
-    JoinHandle,
     message_queue::{TimingEvent, TraceEvent},
-    spawn, turbo_tasks,
+    turbo_tasks,
 };
 
 use crate::database::{
@@ -29,6 +27,8 @@ mod parallel_scheduler;
 /// Number of key families, see KeySpace enum for their numbers.
 const FAMILIES: usize = 4;
 
+const COMPACTION_MESSAGE: &str = "Finished filesystem cache database compaction";
+
 const MB: u64 = 1024 * 1024;
 const COMPACT_CONFIG: CompactConfig = CompactConfig {
     min_merge_count: 3,
@@ -42,7 +42,6 @@ const COMPACT_CONFIG: CompactConfig = CompactConfig {
 
 pub struct TurboKeyValueDatabase {
     db: Arc<TurboPersistence<TurboTasksParallelScheduler, FAMILIES>>,
-    compact_join_handle: Mutex<Option<JoinHandle<Result<()>>>>,
     is_ci: bool,
     is_short_session: bool,
     is_fresh: bool,
@@ -61,7 +60,6 @@ impl TurboKeyValueDatabase {
         let db = Arc::new(TurboPersistence::open_with_config(versioned_path, CONFIG)?);
         Ok(Self {
             db: db.clone(),
-            compact_join_handle: Mutex::new(None),
             is_ci,
             is_short_session,
             is_fresh: db.is_empty(),
@@ -105,41 +103,37 @@ impl KeyValueDatabase for TurboKeyValueDatabase {
         Self: 'l;
 
     fn write_batch(&self) -> Result<Self::ConcurrentWriteBatch<'_>> {
-        // Wait for the compaction to finish
-        if let Some(join_handle) = self.compact_join_handle.lock().take() {
-            join_handle.join()?;
-        }
-        // Start a new write batch
         Ok(TurboWriteBatch {
             batch: self.db.write_batch()?,
             db: &self.db,
-            compact_join_handle: (!self.is_short_session && !self.db.is_empty())
-                .then_some(&self.compact_join_handle),
         })
     }
 
     fn prevent_writes(&self) {}
 
-    fn shutdown(&self) -> Result<()> {
-        // Wait for the compaction to finish
-        if let Some(join_handle) = self.compact_join_handle.lock().take() {
-            join_handle.join()?;
+    fn compact(&self) -> Result<bool> {
+        if self.is_short_session || self.db.is_empty() {
+            return Ok(false);
         }
+        do_compact(
+            &self.db,
+            COMPACTION_MESSAGE,
+            available_parallelism().map_or(4, |c| max(4, c.get() / 2)),
+        )
+    }
+
+    fn shutdown(&self) -> Result<()> {
         // Compact the database on shutdown
         // (Avoid compacting a fresh database since we don't have any usage info yet)
         if !self.is_fresh {
             if self.is_ci {
                 // Fully compact in CI to reduce cache size
-                do_compact(
-                    &self.db,
-                    "Finished filesystem cache database compaction",
-                    usize::MAX,
-                )?;
+                do_compact(&self.db, COMPACTION_MESSAGE, usize::MAX)?;
             } else {
                 // Compact with a reasonable limit in non-CI environments
                 do_compact(
                     &self.db,
-                    "Finished filesystem cache database compaction",
+                    COMPACTION_MESSAGE,
                     available_parallelism().map_or(4, |c| max(4, c.get())),
                 )?;
             }
@@ -153,7 +147,7 @@ fn do_compact(
     db: &TurboPersistence<TurboTasksParallelScheduler, FAMILIES>,
     message: &'static str,
     max_merge_segment_count: usize,
-) -> Result<()> {
+) -> Result<bool> {
     let start = Instant::now();
     // SystemTime for wall-clock timestamps in trace events (Instant has no
     // defined epoch so it can't be used for cross-process trace correlation).
@@ -182,14 +176,13 @@ fn do_compact(
             vec![],
         )));
     }
-    Ok(())
+    Ok(ran)
 }
 
 pub struct TurboWriteBatch<'a> {
     batch:
         turbo_persistence::WriteBatch<WriteBuffer<'static>, TurboTasksParallelScheduler, FAMILIES>,
     db: &'a Arc<TurboPersistence<TurboTasksParallelScheduler, FAMILIES>>,
-    compact_join_handle: Option<&'a Mutex<Option<JoinHandle<Result<()>>>>>,
 }
 
 impl<'a> ConcurrentWriteBatch<'a> for TurboWriteBatch<'a> {
@@ -207,22 +200,7 @@ impl<'a> ConcurrentWriteBatch<'a> for TurboWriteBatch<'a> {
     }
 
     fn commit(self) -> Result<()> {
-        // Commit the write batch
         self.db.commit_write_batch(self.batch)?;
-
-        if let Some(compact_join_handle) = self.compact_join_handle {
-            // Start a new compaction in the background
-            let db = self.db.clone();
-            let handle = spawn(async move {
-                do_compact(
-                    &db,
-                    "Finished filesystem cache database compaction",
-                    available_parallelism().map_or(4, |c| max(4, c.get() / 2)),
-                )
-            });
-            compact_join_handle.lock().replace(handle);
-        }
-
         Ok(())
     }
 

filename: turbopack/crates/turbo-tasks-backend/src/kv_backing_storage.rs
patch: @@ -375,6 +375,10 @@ impl<T: KeyValueDatabase + Send + Sync + 'static> BackingStorageSealed
             .collect::<Result<Vec<_>>>()
     }
 
+    fn compact(&self) -> Result<bool> {
+        self.inner.database.compact()
+    }
+
     fn shutdown(&self) -> Result<()> {
         self.inner.database.shutdown()
     }