diff --git a/Cargo.lock b/Cargo.lock
index d2b1b8abfc7e9..09fbaee11dab2 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -4106,6 +4106,7 @@ dependencies = [
  "tracing-chrome",
  "tracing-subscriber",
  "turbo-tasks",
+ "turbo-tasks-backend",
  "turbo-tasks-build",
  "turbo-tasks-fs",
  "turbo-tasks-malloc",
diff --git a/Cargo.toml b/Cargo.toml
index 9c69dcbeb75df..8f3b5bbf35f95 100644
--- a/Cargo.toml
+++ b/Cargo.toml
@@ -44,6 +44,7 @@ swc-ast-explorer = { path = "turbopack/crates/turbopack-swc-ast-explorer" }
 turbo-prehash = { path = "turbopack/crates/turbo-prehash" }
 turbo-tasks-malloc = { path = "turbopack/crates/turbo-tasks-malloc", default-features = false }
 turbo-tasks = { path = "turbopack/crates/turbo-tasks" }
+turbo-tasks-backend = { path = "turbopack/crates/turbo-tasks-backend" }
 turbo-tasks-build = { path = "turbopack/crates/turbo-tasks-build" }
 turbo-tasks-bytes = { path = "turbopack/crates/turbo-tasks-bytes" }
 turbo-tasks-env = { path = "turbopack/crates/turbo-tasks-env" }
diff --git a/crates/napi/Cargo.toml b/crates/napi/Cargo.toml
index e7d428bbeb4fb..604dd24188771 100644
--- a/crates/napi/Cargo.toml
+++ b/crates/napi/Cargo.toml
@@ -37,6 +37,8 @@ __internal_dhat-heap = ["dhat"]
 # effectively does nothing.
 __internal_dhat-ad-hoc = ["dhat"]
 
+new-backend = ["dep:turbo-tasks-backend"]
+
 # Enable specific tls features per-target.
 [target.'cfg(all(target_os = "windows", target_arch = "aarch64"))'.dependencies]
 next-core = { workspace = true, features = ["native-tls"] }
@@ -105,6 +107,7 @@ lightningcss-napi = { workspace = true }
 tokio = { workspace = true, features = ["full"] }
 turbo-tasks = { workspace = true }
 turbo-tasks-memory = { workspace = true }
+turbo-tasks-backend = { workspace = true, optional = true }
 turbo-tasks-fs = { workspace = true }
 next-api = { workspace = true }
 next-build = { workspace = true }
diff --git a/crates/napi/src/next_api/project.rs b/crates/napi/src/next_api/project.rs
index 6cd5ea8316d02..d23bd930b9d5b 100644
--- a/crates/napi/src/next_api/project.rs
+++ b/crates/napi/src/next_api/project.rs
@@ -1,4 +1,4 @@
-use std::{io::Write, path::PathBuf, sync::Arc, thread, time::Duration};
+use std::{path::PathBuf, sync::Arc, thread, time::Duration};
 
 use anyhow::{anyhow, bail, Context, Result};
 use napi::{
@@ -43,8 +43,8 @@ use url::Url;
 use super::{
     endpoint::ExternalEndpoint,
     utils::{
-        get_diagnostics, get_issues, subscribe, NapiDiagnostic, NapiIssue, NextBackend, RootTask,
-        TurbopackResult, VcArc,
+        create_turbo_tasks, get_diagnostics, get_issues, subscribe, NapiDiagnostic, NapiIssue,
+        NextBackend, RootTask, TurbopackResult, VcArc,
     },
 };
 use crate::register;
@@ -88,7 +88,7 @@ pub struct NapiProjectOptions {
 
     /// next.config's distDir. Project initialization occurs eariler than
     /// deserializing next.config, so passing it as separate option.
-    pub dist_dir: Option<String>,
+    pub dist_dir: String,
 
     /// Whether to watch he filesystem for file changes.
     pub watch: bool,
@@ -279,10 +279,7 @@ pub async fn project_new(
         let subscriber = Registry::default();
 
         let subscriber = subscriber.with(EnvFilter::builder().parse(trace).unwrap());
-        let dist_dir = options
-            .dist_dir
-            .as_ref()
-            .map_or_else(|| ".next".to_string(), |d| d.to_string());
+        let dist_dir = options.dist_dir.clone();
 
         let internal_dir = PathBuf::from(&options.project_path).join(dist_dir);
         std::fs::create_dir_all(&internal_dir)
@@ -308,27 +305,29 @@ pub async fn project_new(
         subscriber.init();
     }
 
-    let turbo_tasks = TurboTasks::new(NextBackend::new(
-        turbo_engine_options
-            .memory_limit
-            .map(|m| m as usize)
-            .unwrap_or(usize::MAX),
-    ));
-    let stats_path = std::env::var_os("NEXT_TURBOPACK_TASK_STATISTICS");
-    if let Some(stats_path) = stats_path {
-        let task_stats = turbo_tasks.backend().task_statistics().enable().clone();
-        exit.on_exit(async move {
-            tokio::task::spawn_blocking(move || {
-                let mut file = std::fs::File::create(&stats_path)
-                    .with_context(|| format!("failed to create or open {stats_path:?}"))?;
-                serde_json::to_writer(&file, &task_stats)
-                    .context("failed to serialize or write task statistics")?;
-                file.flush().context("failed to flush file")
-            })
-            .await
-            .unwrap()
-            .unwrap();
-        });
+    let memory_limit = turbo_engine_options
+        .memory_limit
+        .map(|m| m as usize)
+        .unwrap_or(usize::MAX);
+    let turbo_tasks = create_turbo_tasks(PathBuf::from(&options.dist_dir), memory_limit)?;
+    #[cfg(not(feature = "new-backend"))]
+    {
+        let stats_path = std::env::var_os("NEXT_TURBOPACK_TASK_STATISTICS");
+        if let Some(stats_path) = stats_path {
+            let task_stats = turbo_tasks.backend().task_statistics().enable().clone();
+            exit.on_exit(async move {
+                tokio::task::spawn_blocking(move || {
+                    let mut file = std::fs::File::create(&stats_path)
+                        .with_context(|| format!("failed to create or open {stats_path:?}"))?;
+                    serde_json::to_writer(&file, &task_stats)
+                        .context("failed to serialize or write task statistics")?;
+                    file.flush().context("failed to flush file")
+                })
+                .await
+                .unwrap()
+                .unwrap();
+            });
+        }
     }
     let options = options.into();
     let container = turbo_tasks
diff --git a/crates/napi/src/next_api/utils.rs b/crates/napi/src/next_api/utils.rs
index bd6bd97b0f2ed..5a7f921ce3742 100644
--- a/crates/napi/src/next_api/utils.rs
+++ b/crates/napi/src/next_api/utils.rs
@@ -1,4 +1,4 @@
-use std::{collections::HashMap, future::Future, ops::Deref, sync::Arc};
+use std::{collections::HashMap, future::Future, ops::Deref, path::PathBuf, sync::Arc};
 
 use anyhow::{anyhow, Context, Result};
 use napi::{
@@ -9,7 +9,6 @@ use napi::{
 use serde::Serialize;
 use turbo_tasks::{ReadRef, TaskId, TryJoinIterExt, TurboTasks, Vc};
 use turbo_tasks_fs::FileContent;
-use turbo_tasks_memory::MemoryBackend;
 use turbopack_core::{
     diagnostics::{Diagnostic, DiagnosticContextExt, PlainDiagnostic},
     error::PrettyPrintError,
@@ -17,7 +16,24 @@ use turbopack_core::{
     source_pos::SourcePos,
 };
 
-pub type NextBackend = MemoryBackend;
+#[cfg(not(feature = "new-backend"))]
+pub type NextBackend = turbo_tasks_memory::MemoryBackend;
+#[cfg(feature = "new-backend")]
+pub type NextBackend = turbo_tasks_backend::TurboTasksBackend;
+
+#[allow(unused_variables, reason = "feature-gated")]
+pub fn create_turbo_tasks(
+    output_path: PathBuf,
+    memory_limit: usize,
+) -> Result<Arc<TurboTasks<NextBackend>>> {
+    #[cfg(not(feature = "new-backend"))]
+    let backend = TurboTasks::new(turbo_tasks_memory::MemoryBackend::new(memory_limit));
+    #[cfg(feature = "new-backend")]
+    let backend = TurboTasks::new(turbo_tasks_backend::TurboTasksBackend::new(Arc::new(
+        turbo_tasks_backend::LmdbBackingStorage::new(&output_path.join("cache/turbopack"))?,
+    )));
+    Ok(backend)
+}
 
 /// A helper type to hold both a Vc operation and the TurboTasks root process.
 /// Without this, we'd need to pass both individually all over the place
diff --git a/crates/napi/src/turbotrace.rs b/crates/napi/src/turbotrace.rs
index f3281ad3030d4..fef534c04b550 100644
--- a/crates/napi/src/turbotrace.rs
+++ b/crates/napi/src/turbotrace.rs
@@ -1,4 +1,4 @@
-use std::sync::Arc;
+use std::{path::PathBuf, sync::Arc};
 
 use napi::bindgen_prelude::*;
 use node_file_trace::{start, Args};
@@ -8,29 +8,28 @@ use turbopack::{
     resolve_options_context::ResolveOptionsContext,
 };
 
-use crate::next_api::utils::NextBackend;
+use crate::next_api::utils::{self, NextBackend};
 
 #[napi]
-pub fn create_turbo_tasks(memory_limit: Option<i64>) -> External<Arc<TurboTasks<NextBackend>>> {
-    let turbo_tasks = TurboTasks::new(NextBackend::new(
-        memory_limit.map(|m| m as usize).unwrap_or(usize::MAX),
-    ));
-    External::new_with_size_hint(
-        turbo_tasks,
-        memory_limit.map(|u| u as usize).unwrap_or(usize::MAX),
-    )
+pub fn create_turbo_tasks(
+    output_path: String,
+    memory_limit: Option<i64>,
+) -> External<Arc<TurboTasks<NextBackend>>> {
+    let limit = memory_limit.map(|u| u as usize).unwrap_or(usize::MAX);
+    let turbo_tasks = utils::create_turbo_tasks(PathBuf::from(&output_path), limit)
+        .expect("Failed to create TurboTasks");
+    External::new_with_size_hint(turbo_tasks, limit)
 }
 
 #[napi]
 pub async fn run_turbo_tracing(
     options: Buffer,
-    turbo_tasks: Option<External<Arc<TurboTasks<NextBackend>>>>,
+    turbo_tasks: External<Arc<TurboTasks<NextBackend>>>,
 ) -> napi::Result<Vec<String>> {
     let args: Args = serde_json::from_slice(options.as_ref())?;
-    let turbo_tasks = turbo_tasks.map(|t| t.clone());
     let files = start(
         Arc::new(args),
-        turbo_tasks.as_ref(),
+        turbo_tasks.clone(),
         Some(ModuleOptionsContext {
             ecmascript: EcmascriptOptionsContext {
                 enable_types: true,
diff --git a/packages/next/src/build/collect-build-traces.ts b/packages/next/src/build/collect-build-traces.ts
index 7c926ff1f5613..c89e82fae93d6 100644
--- a/packages/next/src/build/collect-build-traces.ts
+++ b/packages/next/src/build/collect-build-traces.ts
@@ -118,6 +118,7 @@ export async function collectBuildTraces({
       let turbotraceOutputPath: string | undefined
       let turbotraceFiles: string[] | undefined
       turboTasksForTrace = bindings.turbo.createTurboTasks(
+        distDir,
         (config.experimental.turbotrace?.memoryLimit ??
           TURBO_TRACE_DEFAULT_MEMORY_LIMIT) *
           1024 *
diff --git a/packages/next/src/build/index.ts b/packages/next/src/build/index.ts
index 7e73394a3bb66..90b812a391c6d 100644
--- a/packages/next/src/build/index.ts
+++ b/packages/next/src/build/index.ts
@@ -1351,6 +1351,7 @@ export default async function build(
           {
             projectPath: dir,
             rootPath: config.outputFileTracingRoot || dir,
+            distDir,
             nextConfig: config,
             jsConfig: await getTurbopackJsConfig(dir, config),
             watch: false,
diff --git a/packages/next/src/build/swc/index.ts b/packages/next/src/build/swc/index.ts
index 3856aacc10e54..b0d9673d609de 100644
--- a/packages/next/src/build/swc/index.ts
+++ b/packages/next/src/build/swc/index.ts
@@ -412,6 +412,11 @@ export interface ProjectOptions {
    */
   projectPath: string
 
+  /**
+   * The path to the .next directory.
+   */
+  distDir: string
+
   /**
    * The next.config.js contents.
    */
@@ -1547,15 +1552,18 @@ function loadNative(importPath?: string) {
       initHeapProfiler: bindings.initHeapProfiler,
       teardownHeapProfiler: bindings.teardownHeapProfiler,
       turbo: {
-        startTrace: (options = {}, turboTasks: unknown) => {
+        startTrace: (options = {}, turboTasks: { __napi: 'TurboTasks' }) => {
           initHeapProfiler()
           return (customBindings ?? bindings).runTurboTracing(
             toBuffer({ exact: true, ...options }),
             turboTasks
           )
         },
-        createTurboTasks: (memoryLimit?: number): unknown =>
-          bindings.createTurboTasks(memoryLimit),
+        createTurboTasks: (
+          outputPath: string,
+          memoryLimit?: number
+        ): { __napi: 'TurboTasks' } =>
+          bindings.createTurboTasks(outputPath, memoryLimit),
         entrypoints: {
           stream: (
             turboTasks: any,
diff --git a/packages/next/src/server/dev/hot-reloader-turbopack.ts b/packages/next/src/server/dev/hot-reloader-turbopack.ts
index d5ffa3a869078..ea4476326b764 100644
--- a/packages/next/src/server/dev/hot-reloader-turbopack.ts
+++ b/packages/next/src/server/dev/hot-reloader-turbopack.ts
@@ -128,6 +128,7 @@ export async function createHotReloaderTurbopack(
     {
       projectPath: dir,
       rootPath: opts.nextConfig.outputFileTracingRoot || dir,
+      distDir,
       nextConfig: opts.nextConfig,
       jsConfig: await getTurbopackJsConfig(dir, nextConfig),
       watch: true,
diff --git a/test/development/basic/next-rs-api.test.ts b/test/development/basic/next-rs-api.test.ts
index ea03557c3ad8d..e7ee5ec8d7ac6 100644
--- a/test/development/basic/next-rs-api.test.ts
+++ b/test/development/basic/next-rs-api.test.ts
@@ -191,6 +191,12 @@ describe('next.rs api', () => {
     console.log(next.testDir)
     const nextConfig = await loadConfig(PHASE_DEVELOPMENT_SERVER, next.testDir)
     const bindings = await loadBindings()
+    const distDir = path.join(
+      process.env.NEXT_SKIP_ISOLATE
+        ? path.resolve(__dirname, '../../..')
+        : next.testDir,
+      '.next'
+    )
     project = await bindings.turbo.createProject({
       env: {},
       jsConfig: {
@@ -198,6 +204,7 @@ describe('next.rs api', () => {
       },
       nextConfig: nextConfig,
       projectPath: next.testDir,
+      distDir,
       rootPath: process.env.NEXT_SKIP_ISOLATE
         ? path.resolve(__dirname, '../../..')
         : next.testDir,
@@ -208,12 +215,7 @@ describe('next.rs api', () => {
         clientRouterFilters: undefined,
         config: nextConfig,
         dev: true,
-        distDir: path.join(
-          process.env.NEXT_SKIP_ISOLATE
-            ? path.resolve(__dirname, '../../..')
-            : next.testDir,
-          '.next'
-        ),
+        distDir: distDir,
         fetchCacheKeyPrefix: undefined,
         hasRewrites: false,
         middlewareMatchers: undefined,
diff --git a/turbopack/crates/node-file-trace/src/lib.rs b/turbopack/crates/node-file-trace/src/lib.rs
index 47adb8f028463..59012e5e3939c 100644
--- a/turbopack/crates/node-file-trace/src/lib.rs
+++ b/turbopack/crates/node-file-trace/src/lib.rs
@@ -28,7 +28,6 @@ use turbo_tasks::{
 use turbo_tasks_fs::{
     glob::Glob, DirectoryEntry, DiskFileSystem, FileSystem, FileSystemPath, ReadGlobResult,
 };
-use turbo_tasks_memory::MemoryBackend;
 use turbopack::{
     emit_asset, emit_with_completion, module_options::ModuleOptionsContext, rebase::RebasedAsset,
     ModuleAssetContext,
@@ -177,7 +176,7 @@ fn default_output_directory() -> String {
 }
 
 impl Args {
-    fn common(&self) -> &CommonArgs {
+    pub fn common(&self) -> &CommonArgs {
         match self {
             Args::Print { common, .. }
             | Args::Annotate { common, .. }
@@ -310,78 +309,16 @@ fn process_input(dir: &Path, context_directory: &str, input: &[String]) -> Resul
         .collect()
 }
 
-pub async fn start(
+pub async fn start<B: Backend>(
     args: Arc<Args>,
-    turbo_tasks: Option<&Arc<TurboTasks<MemoryBackend>>>,
+    turbo_tasks: Arc<TurboTasks<B>>,
     module_options: Option<ModuleOptionsContext>,
     resolve_options: Option<ResolveOptionsContext>,
 ) -> Result<Vec<RcStr>> {
     register();
-    let &CommonArgs {
-        memory_limit,
-        #[cfg(feature = "persistent_cache")]
-            cache: CacheArgs {
-            ref cache,
-            ref cache_fully,
-        },
-        ..
-    } = args.common();
-    #[cfg(feature = "persistent_cache")]
-    if let Some(cache) = cache {
-        use tokio::time::timeout;
-        use turbo_tasks_memory::MemoryBackendWithPersistedGraph;
-        use turbo_tasks_rocksdb::RocksDbPersistedGraph;
-
-        run(
-            &args,
-            || {
-                let start = Instant::now();
-                let backend = MemoryBackendWithPersistedGraph::new(
-                    RocksDbPersistedGraph::new(cache).unwrap(),
-                );
-                let tt = TurboTasks::new(backend);
-                let elapsed = start.elapsed();
-                println!("restored cache {}", FormatDuration(elapsed));
-                tt
-            },
-            |tt, _, duration| async move {
-                let mut start = Instant::now();
-                if *cache_fully {
-                    tt.wait_background_done().await;
-                    tt.stop_and_wait().await;
-                    let elapsed = start.elapsed();
-                    println!("flushed cache {}", FormatDuration(elapsed));
-                } else {
-                    let background_timeout =
-                        std::cmp::max(duration / 5, Duration::from_millis(100));
-                    let timed_out = timeout(background_timeout, tt.wait_background_done())
-                        .await
-                        .is_err();
-                    tt.stop_and_wait().await;
-                    let elapsed = start.elapsed();
-                    if timed_out {
-                        println!("flushed cache partially {}", FormatDuration(elapsed));
-                    } else {
-                        println!("flushed cache completely {}", FormatDuration(elapsed));
-                    }
-                }
-                start = Instant::now();
-                drop(tt);
-                let elapsed = start.elapsed();
-                println!("writing cache {}", FormatDuration(elapsed));
-            },
-        )
-        .await;
-        return;
-    }
-
     run(
-        args.clone(),
-        || {
-            turbo_tasks.cloned().unwrap_or_else(|| {
-                TurboTasks::new(MemoryBackend::new(memory_limit.unwrap_or(usize::MAX)))
-            })
-        },
+        args,
+        turbo_tasks,
         |_, _, _| async move {},
         module_options,
         resolve_options,
@@ -391,7 +328,7 @@ pub async fn start(
 
 async fn run<B: Backend + 'static, F: Future<Output = ()>>(
     args: Arc<Args>,
-    create_tt: impl Fn() -> Arc<TurboTasks<B>>,
+    tt: Arc<TurboTasks<B>>,
     final_finish: impl FnOnce(Arc<TurboTasks<B>>, TaskId, Duration) -> F,
     module_options: Option<ModuleOptionsContext>,
     resolve_options: Option<ResolveOptionsContext>,
@@ -454,7 +391,6 @@ async fn run<B: Backend + 'static, F: Future<Output = ()>>(
         matches!(&*args, Args::Annotate { .. }) || matches!(&*args, Args::Print { .. });
     let (sender, mut receiver) = channel(1);
     let dir = current_dir().unwrap();
-    let tt = create_tt();
     let module_options = TransientInstance::new(module_options.unwrap_or_default());
     let resolve_options = TransientInstance::new(resolve_options.unwrap_or_default());
     let log_options = TransientInstance::new(LogOptions {
diff --git a/turbopack/crates/node-file-trace/src/main.rs b/turbopack/crates/node-file-trace/src/main.rs
index 2e435166363da..77767706c2df9 100644
--- a/turbopack/crates/node-file-trace/src/main.rs
+++ b/turbopack/crates/node-file-trace/src/main.rs
@@ -5,6 +5,8 @@ use std::sync::Arc;
 use anyhow::Result;
 use clap::Parser;
 use node_file_trace::{start, Args};
+use turbo_tasks::TurboTasks;
+use turbo_tasks_memory::MemoryBackend;
 
 #[global_allocator]
 static ALLOC: turbo_tasks_malloc::TurboMalloc = turbo_tasks_malloc::TurboMalloc;
@@ -15,7 +17,10 @@ async fn main() -> Result<()> {
     console_subscriber::init();
     let args = Arc::new(Args::parse());
     let should_print = matches!(&*args, Args::Print { .. });
-    let result = start(args, None, None, None).await?;
+    let turbo_tasks = TurboTasks::new(MemoryBackend::new(
+        args.common().memory_limit.unwrap_or(usize::MAX),
+    ));
+    let result = start(args, turbo_tasks, None, None).await?;
     if should_print {
         for file in result.iter() {
             println!("{}", file);