diff --git a/Cargo.lock b/Cargo.lock
index aaee10e19711a..eeaa124ca7c02 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -8351,6 +8351,27 @@ dependencies = [
  "unsize",
 ]
 
+[[package]]
+name = "turbo-tasks-backend"
+version = "0.1.0"
+dependencies = [
+ "anyhow",
+ "auto-hash-map",
+ "dashmap",
+ "once_cell",
+ "parking_lot",
+ "rustc-hash",
+ "serde",
+ "smallvec",
+ "tokio",
+ "tracing",
+ "turbo-prehash",
+ "turbo-tasks",
+ "turbo-tasks-build",
+ "turbo-tasks-hash",
+ "turbo-tasks-malloc",
+]
+
 [[package]]
 name = "turbo-tasks-build"
 version = "0.1.0"
diff --git a/turbopack/crates/turbo-tasks-backend/Cargo.toml b/turbopack/crates/turbo-tasks-backend/Cargo.toml
new file mode 100644
index 0000000000000..8b21d629dfd3d
--- /dev/null
+++ b/turbopack/crates/turbo-tasks-backend/Cargo.toml
@@ -0,0 +1,32 @@
+[package]
+name = "turbo-tasks-backend"
+version = "0.1.0"
+description = "TBD"
+license = "MPL-2.0"
+edition = "2021"
+autobenches = false
+
+[lib]
+bench = false
+
+[lints]
+workspace = true
+
+[dependencies]
+anyhow = { workspace = true }
+auto-hash-map = { workspace = true }
+dashmap = { workspace = true }
+once_cell = { workspace = true }
+parking_lot = { workspace = true }
+rustc-hash = { workspace = true }
+serde = { workspace = true }
+smallvec = { workspace = true }
+tokio = { workspace = true }
+tracing = { workspace = true }
+turbo-prehash = { workspace = true }
+turbo-tasks = { workspace = true }
+turbo-tasks-hash = { workspace = true }
+turbo-tasks-malloc = { workspace = true, default-features = false }
+
+[build-dependencies]
+turbo-tasks-build = { workspace = true }
diff --git a/turbopack/crates/turbo-tasks-backend/build.rs b/turbopack/crates/turbo-tasks-backend/build.rs
new file mode 100644
index 0000000000000..1673efed59cce
--- /dev/null
+++ b/turbopack/crates/turbo-tasks-backend/build.rs
@@ -0,0 +1,5 @@
+use turbo_tasks_build::generate_register;
+
+fn main() {
+    generate_register();
+}
diff --git a/turbopack/crates/turbo-tasks-backend/src/backend/mod.rs b/turbopack/crates/turbo-tasks-backend/src/backend/mod.rs
new file mode 100644
index 0000000000000..c4b918a446dee
--- /dev/null
+++ b/turbopack/crates/turbo-tasks-backend/src/backend/mod.rs
@@ -0,0 +1,37 @@
+mod operation;
+mod storage;
+
+use std::{collections::VecDeque, sync::Arc};
+
+use parking_lot::Mutex;
+use turbo_tasks::{backend::CachedTaskType, TaskId};
+
+use self::{operation::Operation, storage::Storage};
+use crate::{
+    data::{CachedDataItem, CachedDataUpdate},
+    utils::{bi_map::BiMap, chunked_vec::ChunkedVec},
+};
+
+pub struct TurboTasksBackend {
+    persisted_task_cache_log: Mutex<ChunkedVec<(Arc<CachedTaskType>, TaskId)>>,
+    task_cache: BiMap<Arc<CachedTaskType>, TaskId>,
+    persisted_storage_log: Mutex<ChunkedVec<CachedDataUpdate>>,
+    storage: Storage<TaskId, CachedDataItem>,
+    operations: Mutex<VecDeque<Box<dyn Operation>>>,
+}
+
+impl TurboTasksBackend {
+    pub fn new() -> Self {
+        Self {
+            persisted_task_cache_log: Mutex::new(ChunkedVec::new()),
+            task_cache: BiMap::new(),
+            persisted_storage_log: Mutex::new(ChunkedVec::new()),
+            storage: Storage::new(),
+            operations: Mutex::new(VecDeque::new()),
+        }
+    }
+
+    fn run_operation(&self, operation: Box<dyn Operation>) {
+        self.operations.lock().push_back(operation);
+    }
+}
diff --git a/turbopack/crates/turbo-tasks-backend/src/backend/operation/mod.rs b/turbopack/crates/turbo-tasks-backend/src/backend/operation/mod.rs
new file mode 100644
index 0000000000000..b3b434511bf3f
--- /dev/null
+++ b/turbopack/crates/turbo-tasks-backend/src/backend/operation/mod.rs
@@ -0,0 +1 @@
+pub trait Operation {}
diff --git a/turbopack/crates/turbo-tasks-backend/src/backend/storage.rs b/turbopack/crates/turbo-tasks-backend/src/backend/storage.rs
new file mode 100644
index 0000000000000..957dabd2db994
--- /dev/null
+++ b/turbopack/crates/turbo-tasks-backend/src/backend/storage.rs
@@ -0,0 +1,34 @@
+use auto_hash_map::AutoMap;
+use dashmap::DashMap;
+use turbo_tasks::Keyed;
+
+enum PersistanceState {
+    /// We know that all state of the object is only in the cache and nothing is
+    /// stored in the persistent cache.
+    CacheOnly,
+    /// We know that some state of the object is stored in the persistent cache.
+    Persisted,
+    /// We have never checked the persistent cache for the state of the object.
+    Unknown,
+}
+
+struct InnerStorage<T: Keyed> {
+    map: AutoMap<T::Key, T::Value>,
+    persistance_state: PersistanceState,
+}
+
+pub struct Storage<K, T: Keyed> {
+    map: DashMap<K, InnerStorage<T>>,
+}
+
+impl<K, T: Keyed> Storage<K, T>
+where
+    K: Eq + std::hash::Hash + Clone,
+    T: Keyed,
+{
+    pub fn new() -> Self {
+        Self {
+            map: DashMap::new(),
+        }
+    }
+}
diff --git a/turbopack/crates/turbo-tasks-backend/src/data.rs b/turbopack/crates/turbo-tasks-backend/src/data.rs
new file mode 100644
index 0000000000000..d4a032e966627
--- /dev/null
+++ b/turbopack/crates/turbo-tasks-backend/src/data.rs
@@ -0,0 +1,184 @@
+use turbo_tasks::{util::SharedError, CellId, SharedReference, TaskId};
+
+#[derive(Debug, Copy, Clone)]
+pub struct CellRef {
+    task: TaskId,
+    cell: CellId,
+}
+
+#[derive(Debug, Copy, Clone)]
+pub enum OutputValue {
+    Cell(CellRef),
+    Output(TaskId),
+    Error,
+}
+impl OutputValue {
+    fn is_transient(&self) -> bool {
+        match self {
+            OutputValue::Cell(cell) => cell.task.is_transient(),
+            OutputValue::Output(task) => task.is_transient(),
+            OutputValue::Error => false,
+        }
+    }
+}
+
+#[derive(Debug, Copy, Clone)]
+pub enum RootType {
+    RootTask,
+    OnceTask,
+    ReadingStronglyConsistent,
+}
+
+#[derive(Debug, Copy, Clone)]
+pub enum InProgressState {
+    Scheduled { clean: bool },
+    InProgress { clean: bool, stale: bool },
+}
+
+#[turbo_tasks::with_key]
+#[derive(Debug, Clone)]
+pub enum CachedDataItem {
+    // Output
+    Output {
+        value: OutputValue,
+    },
+    Collectible {
+        collectible: CellRef,
+        value: (),
+    },
+
+    // State
+    Dirty {
+        value: (),
+    },
+    DirtyWhenPersisted {
+        value: (),
+    },
+
+    // Children
+    Child {
+        task: TaskId,
+        value: (),
+    },
+
+    // Cells
+    CellData {
+        cell: CellId,
+        value: SharedReference,
+    },
+
+    // Dependencies
+    OutputDependency {
+        target: TaskId,
+        value: (),
+    },
+    CellDependency {
+        target: CellRef,
+        value: (),
+    },
+
+    // Dependent
+    OutputDependent {
+        task: TaskId,
+        value: (),
+    },
+    CellDependent {
+        cell: CellId,
+        task: TaskId,
+        value: (),
+    },
+
+    // Aggregation Graph
+    AggregationNumber {
+        value: u32,
+    },
+    Follower {
+        task: TaskId,
+        value: (),
+    },
+    Upper {
+        task: TaskId,
+        value: (),
+    },
+
+    // Aggregated Data
+    AggregatedDirtyTask {
+        task: TaskId,
+        value: (),
+    },
+    AggregatedCollectible {
+        collectible: CellRef,
+        value: (),
+    },
+    AggregatedUnfinishedTasks {
+        value: u32,
+    },
+
+    // Transient Root Type
+    AggregateRootType {
+        value: RootType,
+    },
+
+    // Transient In Progress state
+    InProgress {
+        value: InProgressState,
+    },
+    OutdatedCollectible {
+        collectible: CellRef,
+        value: (),
+    },
+    OutdatedOutputDependency {
+        target: TaskId,
+        value: (),
+    },
+    OutdatedCellDependency {
+        target: CellRef,
+        value: (),
+    },
+
+    // Transient Error State
+    Error {
+        value: SharedError,
+    },
+}
+
+impl CachedDataItem {
+    pub fn is_persistent(&self) -> bool {
+        match self {
+            CachedDataItem::Output { value } => value.is_transient(),
+            CachedDataItem::Collectible { collectible, .. } => !collectible.task.is_transient(),
+            CachedDataItem::Dirty { .. } => true,
+            CachedDataItem::DirtyWhenPersisted { .. } => true,
+            CachedDataItem::Child { task, .. } => !task.is_transient(),
+            CachedDataItem::CellData { .. } => true,
+            CachedDataItem::OutputDependency { target, .. } => !target.is_transient(),
+            CachedDataItem::CellDependency { target, .. } => !target.task.is_transient(),
+            CachedDataItem::OutputDependent { task, .. } => !task.is_transient(),
+            CachedDataItem::CellDependent { task, .. } => !task.is_transient(),
+            CachedDataItem::AggregationNumber { .. } => true,
+            CachedDataItem::Follower { task, .. } => !task.is_transient(),
+            CachedDataItem::Upper { task, .. } => !task.is_transient(),
+            CachedDataItem::AggregatedDirtyTask { task, .. } => !task.is_transient(),
+            CachedDataItem::AggregatedCollectible { collectible, .. } => {
+                !collectible.task.is_transient()
+            }
+            CachedDataItem::AggregatedUnfinishedTasks { .. } => true,
+            CachedDataItem::AggregateRootType { .. } => false,
+            CachedDataItem::InProgress { .. } => false,
+            CachedDataItem::OutdatedCollectible { .. } => false,
+            CachedDataItem::OutdatedOutputDependency { .. } => false,
+            CachedDataItem::OutdatedCellDependency { .. } => false,
+            CachedDataItem::Error { .. } => false,
+        }
+    }
+}
+
+trait IsDefault {
+    fn is_default(&self) -> bool;
+}
+
+pub struct CachedDataUpdate {
+    pub task: TaskId,
+    pub key: CachedDataItemKey,
+    pub value: Option<CachedDataItemValue>,
+}
diff --git a/turbopack/crates/turbo-tasks-backend/src/lib.rs b/turbopack/crates/turbo-tasks-backend/src/lib.rs
new file mode 100644
index 0000000000000..5f67a0c86f697
--- /dev/null
+++ b/turbopack/crates/turbo-tasks-backend/src/lib.rs
@@ -0,0 +1,3 @@
+mod backend;
+mod data;
+mod utils;
diff --git a/turbopack/crates/turbo-tasks-backend/src/utils/bi_map.rs b/turbopack/crates/turbo-tasks-backend/src/utils/bi_map.rs
new file mode 100644
index 0000000000000..97392f52b8196
--- /dev/null
+++ b/turbopack/crates/turbo-tasks-backend/src/utils/bi_map.rs
@@ -0,0 +1,50 @@
+use std::{borrow::Borrow, hash::Hash};
+
+use dashmap::{mapref::entry::Entry, DashMap};
+
+pub struct BiMap<K, V> {
+    forward: DashMap<K, V>,
+    reverse: DashMap<V, K>,
+}
+
+impl<K, V> BiMap<K, V>
+where
+    K: Eq + Hash + Clone,
+    V: Eq + Hash + Clone,
+{
+    pub fn new() -> Self {
+        Self {
+            forward: DashMap::new(),
+            reverse: DashMap::new(),
+        }
+    }
+
+    pub fn lookup_forward<Q>(&self, key: &Q) -> Option<V>
+    where
+        K: Borrow<Q>,
+        Q: Hash + Eq,
+    {
+        self.forward.get(key).map(|v| v.value().clone())
+    }
+
+    pub fn lookup_reverse<Q>(&self, key: &Q) -> Option<K>
+    where
+        V: Borrow<Q>,
+        Q: Hash + Eq,
+    {
+        self.reverse.get(key).map(|v| v.value().clone())
+    }
+
+    pub fn try_insert(&self, key: K, value: V) -> Result<(), V> {
+        match self.forward.entry(key) {
+            Entry::Occupied(e) => Err(e.get().clone()),
+            Entry::Vacant(e) => {
+                let e = e.insert_entry(value.clone());
+                let key = e.key().clone();
+                drop(e);
+                self.reverse.insert(value, key);
+                Ok(())
+            }
+        }
+    }
+}
diff --git a/turbopack/crates/turbo-tasks-backend/src/utils/chunked_vec.rs b/turbopack/crates/turbo-tasks-backend/src/utils/chunked_vec.rs
new file mode 100644
index 0000000000000..4ac560aab9c9e
--- /dev/null
+++ b/turbopack/crates/turbo-tasks-backend/src/utils/chunked_vec.rs
@@ -0,0 +1,76 @@
+pub struct ChunkedVec<T> {
+    chunks: Vec<Vec<T>>,
+}
+
+impl<T> ChunkedVec<T> {
+    pub fn new() -> Self {
+        Self { chunks: Vec::new() }
+    }
+
+    pub fn len(&self) -> usize {
+        if let Some(last) = self.chunks.last() {
+            let free = last.capacity() - self.len();
+            cummulative_chunk_size(self.chunks.len() - 1) - free
+        } else {
+            0
+        }
+    }
+
+    pub fn push(&mut self, item: T) {
+        if let Some(chunk) = self.chunks.last_mut() {
+            if chunk.len() < chunk.capacity() {
+                chunk.push(item);
+                return;
+            }
+        }
+        let mut chunk = Vec::with_capacity(chunk_size(self.chunks.len()));
+        chunk.push(item);
+        self.chunks.push(chunk);
+    }
+
+    pub fn into_iter(self) -> impl Iterator<Item = T> {
+        let len = self.len();
+        ExactSizeIter {
+            iter: self.chunks.into_iter().flat_map(|chunk| chunk.into_iter()),
+            len,
+        }
+    }
+
+    pub fn iter(&self) -> impl Iterator<Item = &T> {
+        ExactSizeIter {
+            iter: self.chunks.iter().flat_map(|chunk| chunk.iter()),
+            len: self.len(),
+        }
+    }
+}
+
+fn chunk_size(chunk_index: usize) -> usize {
+    8 << chunk_index
+}
+
+fn cummulative_chunk_size(chunk_index: usize) -> usize {
+    8 << (chunk_index + 1) - 8
+}
+
+struct ExactSizeIter<I: Iterator> {
+    iter: I,
+    len: usize,
+}
+
+impl<I: Iterator> Iterator for ExactSizeIter<I> {
+    type Item = I::Item;
+
+    fn next(&mut self) -> Option<Self::Item> {
+        self.iter.next().inspect(|_| self.len -= 1)
+    }
+
+    fn size_hint(&self) -> (usize, Option<usize>) {
+        (self.len, Some(self.len))
+    }
+}
+
+impl<I: Iterator> ExactSizeIterator for ExactSizeIter<I> {
+    fn len(&self) -> usize {
+        self.len
+    }
+}
diff --git a/turbopack/crates/turbo-tasks-backend/src/utils/mod.rs b/turbopack/crates/turbo-tasks-backend/src/utils/mod.rs
new file mode 100644
index 0000000000000..54b7d067e843b
--- /dev/null
+++ b/turbopack/crates/turbo-tasks-backend/src/utils/mod.rs
@@ -0,0 +1,2 @@
+pub mod bi_map;
+pub mod chunked_vec;
diff --git a/turbopack/crates/turbo-tasks-macros/src/lib.rs b/turbopack/crates/turbo-tasks-macros/src/lib.rs
index 75124c7ae8c0e..217c4dd54c299 100644
--- a/turbopack/crates/turbo-tasks-macros/src/lib.rs
+++ b/turbopack/crates/turbo-tasks-macros/src/lib.rs
@@ -11,6 +11,7 @@ mod primitive_macro;
 mod value_impl_macro;
 mod value_macro;
 mod value_trait_macro;
+mod with_key_macro;
 
 extern crate proc_macro;
 
@@ -177,6 +178,12 @@ pub fn value_impl(args: TokenStream, input: TokenStream) -> TokenStream {
     value_impl_macro::value_impl(args, input)
 }
 
+#[proc_macro_error]
+#[proc_macro_attribute]
+pub fn with_key(_args: TokenStream, input: TokenStream) -> TokenStream {
+    with_key_macro::with_key(input)
+}
+
 #[allow_internal_unstable(min_specialization, into_future, trivial_bounds)]
 #[proc_macro_error]
 #[proc_macro]
diff --git a/turbopack/crates/turbo-tasks-macros/src/with_key_macro.rs b/turbopack/crates/turbo-tasks-macros/src/with_key_macro.rs
new file mode 100644
index 0000000000000..996ad6b8dc3c7
--- /dev/null
+++ b/turbopack/crates/turbo-tasks-macros/src/with_key_macro.rs
@@ -0,0 +1,179 @@
+use proc_macro::TokenStream;
+use quote::quote;
+use syn::{parse_macro_input, Ident, ItemEnum};
+
+pub fn with_key(input: TokenStream) -> TokenStream {
+    let input = parse_macro_input!(input as ItemEnum);
+
+    let attrs = &input.attrs;
+    let ident = &input.ident;
+    let vis = &input.vis;
+    let key_name = Ident::new(&format!("{}Key", input.ident), input.ident.span());
+    let value_name = Ident::new(&format!("{}Value", input.ident), input.ident.span());
+
+    let variant_names = input
+        .variants
+        .iter()
+        .map(|variant| &variant.ident)
+        .collect::<Vec<_>>();
+
+    let key_fields = input
+        .variants
+        .iter()
+        .map(|variant| {
+            variant
+                .fields
+                .iter()
+                .filter(|field| {
+                    let Some(ident) = &field.ident else {
+                        return false;
+                    };
+                    ident != "value"
+                })
+                .collect::<Vec<_>>()
+        })
+        .collect::<Vec<_>>();
+
+    let value_fields = input
+        .variants
+        .iter()
+        .map(|variant| {
+            variant
+                .fields
+                .iter()
+                .filter(|field| {
+                    let Some(ident) = &field.ident else {
+                        return false;
+                    };
+                    ident == "value"
+                })
+                .collect::<Vec<_>>()
+        })
+        .collect::<Vec<_>>();
+
+    let key_decl = field_declarations(&key_fields);
+    let key_pat = patterns(&key_fields);
+    let key_clone_fields = clone_fields(&key_fields);
+
+    let value_decl = field_declarations(&value_fields);
+    let value_pat = patterns(&value_fields);
+    let value_clone_fields = clone_fields(&value_fields);
+
+    quote! {
+        #input
+
+        impl turbo_tasks::Keyed for #ident {
+            type Key = #key_name;
+            type Value = #value_name;
+
+            fn key(&self) -> #key_name {
+                match self {
+                    #(
+                        #ident::#variant_names { #key_pat .. } => #key_name::#variant_names { #key_clone_fields },
+                    )*
+                }
+            }
+
+            fn value(&self) -> #value_name {
+                match self {
+                    #(
+                        #ident::#variant_names { #value_pat .. } => #value_name::#variant_names { #value_clone_fields },
+                    )*
+                }
+            }
+
+            fn from_key_and_value(key: #key_name, value: #value_name) -> Self {
+                match (key, value) {
+                    #(
+                        (#key_name::#variant_names { #key_pat }, #value_name::#variant_names { #value_pat }) => #ident::#variant_names { #key_pat #value_pat },
+                    )*
+                    _ => panic!("Invalid key and value combination"),
+                }
+            }
+        }
+
+        #(#attrs)*
+        #vis enum #key_name {
+            #(
+                #variant_names {
+                    #key_decl
+                },
+            )*
+        }
+
+        #(#attrs)*
+        #vis enum #value_name {
+            #(
+                #variant_names {
+                    #value_decl
+                },
+            )*
+        }
+    }
+    .into()
+}
+
+fn patterns(fields: &Vec<Vec<&syn::Field>>) -> Vec<proc_macro2::TokenStream> {
+    let variant_pat = fields
+        .iter()
+        .map(|fields| {
+            let pat = fields
+                .iter()
+                .map(|field| {
+                    let ident = field.ident.as_ref().unwrap();
+                    quote! {
+                        #ident
+                    }
+                })
+                .collect::<Vec<_>>();
+            quote! {
+                #(#pat,)*
+            }
+        })
+        .collect::<Vec<_>>();
+    variant_pat
+}
+
+fn clone_fields(fields: &Vec<Vec<&syn::Field>>) -> Vec<proc_macro2::TokenStream> {
+    let variant_pat = fields
+        .iter()
+        .map(|fields| {
+            let pat = fields
+                .iter()
+                .map(|field| {
+                    let ident = field.ident.as_ref().unwrap();
+                    quote! {
+                        #ident: #ident.clone()
+                    }
+                })
+                .collect::<Vec<_>>();
+            quote! {
+                #(#pat,)*
+            }
+        })
+        .collect::<Vec<_>>();
+    variant_pat
+}
+
+fn field_declarations(fields: &Vec<Vec<&syn::Field>>) -> Vec<proc_macro2::TokenStream> {
+    fields
+        .iter()
+        .map(|fields| {
+            let fields = fields
+                .iter()
+                .map(|field| {
+                    let ty = &field.ty;
+                    let ident = field.ident.as_ref().unwrap();
+                    let attrs = &field.attrs;
+                    quote! {
+                        #(#attrs)*
+                        #ident: #ty
+                    }
+                })
+                .collect::<Vec<_>>();
+            quote! {
+                #(#fields),*
+            }
+        })
+        .collect::<Vec<_>>()
+}
diff --git a/turbopack/crates/turbo-tasks/src/keyed.rs b/turbopack/crates/turbo-tasks/src/keyed.rs
new file mode 100644
index 0000000000000..46a1cf7f6b9bf
--- /dev/null
+++ b/turbopack/crates/turbo-tasks/src/keyed.rs
@@ -0,0 +1,7 @@
+pub trait Keyed {
+    type Key;
+    type Value;
+    fn key(&self) -> Self::Key;
+    fn value(&self) -> Self::Value;
+    fn from_key_and_value(key: Self::Key, value: Self::Value) -> Self;
+}
diff --git a/turbopack/crates/turbo-tasks/src/lib.rs b/turbopack/crates/turbo-tasks/src/lib.rs
index 1d44af013541c..70fb61a46c8ee 100644
--- a/turbopack/crates/turbo-tasks/src/lib.rs
+++ b/turbopack/crates/turbo-tasks/src/lib.rs
@@ -49,6 +49,7 @@ mod id;
 mod id_factory;
 mod invalidation;
 mod join_iter_ext;
+mod keyed;
 #[doc(hidden)]
 pub mod macro_helpers;
 mod magic_any;
@@ -86,6 +87,7 @@ pub use invalidation::{
     DynamicEqHash, InvalidationReason, InvalidationReasonKind, InvalidationReasonSet,
 };
 pub use join_iter_ext::{JoinIterExt, TryFlatJoinIterExt, TryJoinIterExt};
+pub use keyed::Keyed;
 pub use magic_any::MagicAny;
 pub use manager::{
     dynamic_call, dynamic_this_call, emit, get_invalidator, mark_finished, mark_stateful,
@@ -101,7 +103,7 @@ use rustc_hash::FxHasher;
 pub use state::State;
 pub use task::{task_input::TaskInput, SharedReference};
 pub use trait_ref::{IntoTraitRef, TraitRef};
-pub use turbo_tasks_macros::{function, value, value_impl, value_trait, TaskInput};
+pub use turbo_tasks_macros::{function, value, value_impl, value_trait, with_key, TaskInput};
 pub use value::{TransientInstance, TransientValue, Value};
 pub use value_type::{TraitMethod, TraitType, ValueType};
 pub use vc::{