diff --git a/foyer-bench/src/main.rs b/foyer-bench/src/main.rs
index 431494b1..6113dac7 100644
--- a/foyer-bench/src/main.rs
+++ b/foyer-bench/src/main.rs
@@ -40,10 +40,9 @@ use exporter::PrometheusExporter;
 #[cfg(target_os = "linux")]
 use foyer::UringIoEngineConfig;
 use foyer::{
-    BlockEngineConfig, Code, Compression, Device, DeviceBuilder, EngineConfig, FifoConfig, FifoPicker,
-    FileDeviceBuilder, FsDeviceBuilder, HybridCache, HybridCacheBuilder, HybridCachePolicy, HybridCacheProperties,
-    InvalidRatioPicker, IoEngineConfig, LfuConfig, LruConfig, NoopDeviceBuilder, PsyncIoEngineConfig, RecoverMode,
-    S3FifoConfig, Spawner, Throttle, TracingOptions,
+    BlockEngineConfig, Code, Compression, Device, DeviceBuilder, FifoConfig, FifoPicker, FileDeviceBuilder,
+    FsDeviceBuilder, HybridCache, HybridCacheBuilder, HybridCachePolicy, InvalidRatioPicker, IoEngineConfig, LfuConfig,
+    LruConfig, NoopDeviceBuilder, PsyncIoEngineConfig, RecoverMode, S3FifoConfig, Spawner, Throttle, TracingOptions,
 };
 use futures_util::future::join_all;
 use itertools::Itertools;
@@ -606,7 +605,7 @@ async fn benchmark(args: Args) {
         _ => unreachable!(),
     };
 
-    let engine_config: Box<dyn EngineConfig<u64, Value, HybridCacheProperties>> = {
+    let engine_config = {
         let mut builder = BlockEngineConfig::new(device)
             .with_block_size(args.block_size.as_u64() as _)
             .with_indexer_shards(args.shards)
@@ -623,8 +622,7 @@ async fn benchmark(args: Args) {
             builder = builder.with_clean_block_threshold(args.clean_block_threshold);
         }
         builder
-    }
-    .boxed();
+    };
 
     let mut builder = builder
         .with_weighter(|_: &u64, value: &Value| u64::BITS as usize / 8 + value.len())
diff --git a/foyer-storage/src/engine/block/engine.rs b/foyer-storage/src/engine/block/engine.rs
index 2e540a8a..1dc42556 100644
--- a/foyer-storage/src/engine/block/engine.rs
+++ b/foyer-storage/src/engine/block/engine.rs
@@ -15,6 +15,7 @@
 use std::{
     fmt::Debug,
     future::Future,
+    hash::Hash,
     marker::PhantomData,
     sync::{
         Arc,
@@ -23,6 +24,7 @@ use std::{
     time::Instant,
 };
 
+use equivalent::Equivalent;
 #[cfg(feature = "tracing")]
 use fastrace::prelude::*;
 use foyer_common::{
@@ -460,19 +462,10 @@ where
     V: StorageValue,
     P: Properties,
 {
-    fn build(self: Box<Self>, ctx: EngineBuildContext) -> BoxFuture<'static, Result<Arc<dyn Engine<K, V, P>>>> {
-        async move { self.build(ctx).await.map(|e| e as Arc<dyn Engine<K, V, P>>) }.boxed()
-    }
-}
+    type Engine = BlockEngine<K, V, P>;
 
-impl<K, V, P> From<BlockEngineConfig<K, V, P>> for Box<dyn EngineConfig<K, V, P>>
-where
-    K: StorageKey,
-    V: StorageValue,
-    P: Properties,
-{
-    fn from(builder: BlockEngineConfig<K, V, P>) -> Self {
-        builder.boxed()
+    fn build(self: Box<Self>, ctx: EngineBuildContext) -> BoxFuture<'static, Result<Arc<BlockEngine<K, V, P>>>> {
+        async move { self.build(ctx).await }.boxed()
     }
 }
 
@@ -549,23 +542,18 @@ where
     V: StorageValue,
     P: Properties,
 {
-    fn wait(&self) -> impl Future<Output = ()> + Send + 'static {
-        let flushers = self.inner.flushers.clone();
-        let block_manager = self.inner.block_manager.clone();
+    fn wait(&self) -> impl Future<Output = ()> + Send + '_ {
+        let inner = &self.inner;
         async move {
-            join_all(flushers.iter().map(|flusher| flusher.wait())).await;
-            block_manager.wait_reclaim().await;
+            join_all(inner.flushers.iter().map(|flusher| flusher.wait())).await;
+            inner.block_manager.wait_reclaim().await;
         }
     }
 
-    fn close(&self) -> BoxFuture<'static, Result<()>> {
-        let this = self.clone();
-        async move {
-            this.inner.active.store(false, Ordering::Relaxed);
-            this.wait().await;
-            Ok(())
-        }
-        .boxed()
+    async fn close(&self) -> Result<()> {
+        self.inner.active.store(false, Ordering::Relaxed);
+        self.wait().await;
+        Ok(())
     }
 
     #[cfg_attr(feature = "tracing", trace(name = "foyer::storage::engine::block::engine::enqueue"))]
@@ -603,21 +591,22 @@ where
         });
     }
 
-    fn load(&self, hash: u64) -> impl Future<Output = Result<Load<K, V, P>>> + Send + 'static {
+    fn load<'a, Q>(&'a self, hash: u64, key: &'a Q) -> impl Future<Output = Result<Load<K, V, P>>> + Send + 'a
+    where
+        Q: Hash + Equivalent<K> + Sync + ?Sized,
+    {
         tracing::trace!(hash, "[block engine]: load");
 
         #[cfg(any(test, feature = "test_utils"))]
         let load_holer = self.inner.load_holder.wait();
 
-        let indexer = self.inner.indexer.clone();
-        let metrics = self.inner.metrics.clone();
-        let block_manager = self.inner.block_manager.clone();
+        let inner = &self.inner;
 
         let load = async move {
             #[cfg(any(test, feature = "test_utils"))]
             load_holer.await;
 
-            let addr = match indexer.get(hash) {
+            let addr = match inner.indexer.get(hash) {
                 Some(addr) => addr,
                 None => {
                     return Ok(Load::Miss);
@@ -626,7 +615,7 @@ where
 
             tracing::trace!(hash, ?addr, "[block engine]: load");
 
-            let block = block_manager.block(addr.block);
+            let block = inner.block_manager.block(addr.block);
             if block.partition().statistics().is_read_throttled() {
                 return Ok(Load::Throttled);
             }
@@ -655,7 +644,7 @@ where
                                 ?e,
                                 "[block engine load]: deserialize read buffer raise error, remove this entry and skip"
                             );
-                            indexer.remove(hash);
+                            inner.indexer.remove(hash);
                             Ok(Load::Miss)
                         }
                         _ => {
@@ -666,7 +655,7 @@ where
                 }
             };
 
-            let (key, value) = {
+            let (k, value) = {
                 let now = Instant::now();
                 let res = match EntryDeserializer::deserialize::<K, V>(
                     &buf[EntryHeader::serialized_len()..],
@@ -686,7 +675,7 @@ where
                                     ?e,
                                     "[block engine load]: deserialize read buffer raise error, remove this entry and skip"
                                 );
-                                indexer.remove(hash);
+                                inner.indexer.remove(hash);
                                 Ok(Load::Miss)
                             }
                             _ => {
@@ -696,19 +685,25 @@ where
                         };
                     }
                 };
-                metrics
+                inner
+                    .metrics
                     .storage_entry_deserialize_duration
                     .record(now.elapsed().as_secs_f64());
                 res
             };
 
+            if !key.equivalent(&k) {
+                inner.metrics.storage_false_positive.increase(1);
+                return Ok(Load::Miss);
+            }
+
             let age = match block.statistics().probation.load(Ordering::Relaxed) {
                 true => Age::Old,
                 false => Age::Young,
             };
 
             Ok(Load::Entry {
-                key,
+                key: k,
                 value,
                 populated: Populated { age },
             })
@@ -720,7 +715,10 @@ where
         load
     }
 
-    fn delete(&self, hash: u64) {
+    fn delete<Q>(&self, hash: u64, _key: &Q)
+    where
+        Q: Hash + Equivalent<K> + Sync + ?Sized,
+    {
         if !self.inner.active.load(Ordering::Relaxed) {
             tracing::warn!("cannot delete entry after closed");
             return;
@@ -744,53 +742,54 @@ where
         });
     }
 
-    fn may_contains(&self, hash: u64) -> bool {
+    fn may_contains<Q>(&self, hash: u64, _key: &Q) -> bool
+    where
+        Q: Hash + Equivalent<K> + Sync + ?Sized,
+    {
         self.inner.indexer.get(hash).is_some()
     }
 
-    fn destroy(&self) -> BoxFuture<'static, Result<()>> {
-        let this = self.clone();
-        async move {
-            if !this.inner.active.load(Ordering::Relaxed) {
-                return Err(Error::new(ErrorKind::Closed, "cannot delete entry after closed"));
-            }
+    async fn destroy(&self) -> Result<()> {
+        if !self.inner.active.load(Ordering::Relaxed) {
+            return Err(Error::new(ErrorKind::Closed, "cannot delete entry after closed"));
+        }
 
-            // Write a tombstone to clear tombstone log by increase the max sequence.
-            let sequence = this.inner.sequence.fetch_add(1, Ordering::Relaxed);
+        // Write a tombstone to clear tombstone log by increase the max sequence.
+        let sequence = self.inner.sequence.fetch_add(1, Ordering::Relaxed);
 
-            this.inner.flushers[0].submit(Submission::Tombstone {
-                tombstone: Tombstone { hash: 0, sequence },
-                stats: None,
-            });
-            this.wait().await;
-
-            // Clear indices.
-            //
-            // This step must perform after the latest writer finished,
-            // otherwise the indices of the latest batch cannot be cleared.
-            this.inner.indexer.clear();
-
-            // Clean blocks.
-            try_join_all((0..this.inner.block_manager.blocks() as BlockId).map(|id| {
-                let block = this.inner.block_manager.block(id).clone();
-                async move {
-                    let res = BlockCleaner::clean(&block).await;
-                    block.statistics().reset();
-                    res
-                }
-            }))
-            .await?;
+        self.inner.flushers[0].submit(Submission::Tombstone {
+            tombstone: Tombstone { hash: 0, sequence },
+            stats: None,
+        });
+        self.wait().await;
+
+        // Clear indices.
+        //
+        // This step must perform after the latest writer finished,
+        // otherwise the indices of the latest batch cannot be cleared.
+        self.inner.indexer.clear();
+
+        // Clean blocks.
+        try_join_all((0..self.inner.block_manager.blocks() as BlockId).map(|id| {
+            let block = self.inner.block_manager.block(id).clone();
+            async move {
+                let res = BlockCleaner::clean(&block).await;
+                block.statistics().reset();
+                res
+            }
+        }))
+        .await?;
 
-            Ok(())
-        }
-        .boxed()
+        Ok(())
     }
 
+    /// Hold flush operations for testing.
     #[cfg(any(test, feature = "test_utils"))]
     pub fn hold_flush(&self) {
         self.inner.flush_switch.on();
     }
 
+    /// Unhold flush operations for testing.
     #[cfg(any(test, feature = "test_utils"))]
     pub fn unhold_flush(&self) {
         self.inner.flush_switch.off();
@@ -817,29 +816,36 @@ where
         self.enqueue(piece, estimated_size);
     }
 
-    fn load(&self, hash: u64) -> BoxFuture<'static, Result<Load<K, V, P>>> {
-        // TODO(MrCroxx): refactor this.
-        self.load(hash).boxed()
+    fn load<'a, Q>(&'a self, hash: u64, key: &'a Q) -> impl Future<Output = Result<Load<K, V, P>>> + Send + 'a
+    where
+        Q: Hash + Equivalent<K> + Sync + ?Sized,
+    {
+        self.load(hash, key)
     }
 
-    fn delete(&self, hash: u64) {
-        self.delete(hash);
+    fn delete<Q>(&self, hash: u64, key: &Q)
+    where
+        Q: Hash + Equivalent<K> + Sync + ?Sized,
+    {
+        self.delete(hash, key);
     }
 
-    fn may_contains(&self, hash: u64) -> bool {
-        self.may_contains(hash)
+    fn may_contains<Q>(&self, hash: u64, key: &Q) -> bool
+    where
+        Q: Hash + Equivalent<K> + Sync + ?Sized,
+    {
+        self.may_contains(hash, key)
     }
 
-    fn destroy(&self) -> BoxFuture<'static, Result<()>> {
+    fn destroy(&self) -> impl Future<Output = Result<()>> + Send + '_ {
         self.destroy()
     }
 
-    fn wait(&self) -> BoxFuture<'static, ()> {
-        // TODO(MrCroxx): refactor this.
-        self.wait().boxed()
+    fn wait(&self) -> impl Future<Output = ()> + Send + '_ {
+        self.wait()
     }
 
-    fn close(&self) -> BoxFuture<'static, Result<()>> {
+    fn close(&self) -> impl Future<Output = Result<()>> + Send + '_ {
         self.close()
     }
 }
@@ -1000,9 +1006,9 @@ mod tests {
         store.unhold_flush();
         store.wait().await;
 
-        let r1 = store.load(memory.hash(&1)).await.unwrap().kv().unwrap();
+        let r1 = store.load(memory.hash(&1), &1).await.unwrap().kv().unwrap();
         assert_eq!(r1, (1, vec![1; 7 * KB]));
-        let r2 = store.load(memory.hash(&2)).await.unwrap().kv().unwrap();
+        let r2 = store.load(memory.hash(&2), &2).await.unwrap().kv().unwrap();
         assert_eq!(r2, (2, vec![2; 3 * KB]));
 
         // [ [e1, e2], [e3, e4], [], [] ]
@@ -1014,13 +1020,13 @@ mod tests {
         store.unhold_flush();
         store.wait().await;
 
-        let r1 = store.load(memory.hash(&1)).await.unwrap().kv().unwrap();
+        let r1 = store.load(memory.hash(&1), &1).await.unwrap().kv().unwrap();
         assert_eq!(r1, (1, vec![1; 7 * KB]));
-        let r2 = store.load(memory.hash(&2)).await.unwrap().kv().unwrap();
+        let r2 = store.load(memory.hash(&2), &2).await.unwrap().kv().unwrap();
         assert_eq!(r2, (2, vec![2; 3 * KB]));
-        let r3 = store.load(memory.hash(&3)).await.unwrap().kv().unwrap();
+        let r3 = store.load(memory.hash(&3), &3).await.unwrap().kv().unwrap();
         assert_eq!(r3, (3, vec![3; 7 * KB]));
-        let r4 = store.load(memory.hash(&4)).await.unwrap().kv().unwrap();
+        let r4 = store.load(memory.hash(&4), &4).await.unwrap().kv().unwrap();
         assert_eq!(r4, (4, vec![4; 2 * KB]));
 
         // [ [e1, e2], [e3, e4], [e5], [] ]
@@ -1028,15 +1034,15 @@ mod tests {
         enqueue(&store, e5);
         store.wait().await;
 
-        let r1 = store.load(memory.hash(&1)).await.unwrap().kv().unwrap();
+        let r1 = store.load(memory.hash(&1), &1).await.unwrap().kv().unwrap();
         assert_eq!(r1, (1, vec![1; 7 * KB]));
-        let r2 = store.load(memory.hash(&2)).await.unwrap().kv().unwrap();
+        let r2 = store.load(memory.hash(&2), &2).await.unwrap().kv().unwrap();
         assert_eq!(r2, (2, vec![2; 3 * KB]));
-        let r3 = store.load(memory.hash(&3)).await.unwrap().kv().unwrap();
+        let r3 = store.load(memory.hash(&3), &3).await.unwrap().kv().unwrap();
         assert_eq!(r3, (3, vec![3; 7 * KB]));
-        let r4 = store.load(memory.hash(&4)).await.unwrap().kv().unwrap();
+        let r4 = store.load(memory.hash(&4), &4).await.unwrap().kv().unwrap();
         assert_eq!(r4, (4, vec![4; 2 * KB]));
-        let r5 = store.load(memory.hash(&5)).await.unwrap().kv().unwrap();
+        let r5 = store.load(memory.hash(&5), &5).await.unwrap().kv().unwrap();
         assert_eq!(r5, (5, vec![5; 11 * KB]));
 
         // [ [], [e3, e4], [e5], [e6, e4*] ]
@@ -1048,15 +1054,15 @@ mod tests {
         store.unhold_flush();
         store.wait().await;
 
-        assert!(store.load(memory.hash(&1)).await.unwrap().kv().is_none());
-        assert!(store.load(memory.hash(&2)).await.unwrap().kv().is_none());
-        let r3 = store.load(memory.hash(&3)).await.unwrap().kv().unwrap();
+        assert!(store.load(memory.hash(&1), &1).await.unwrap().kv().is_none());
+        assert!(store.load(memory.hash(&2), &2).await.unwrap().kv().is_none());
+        let r3 = store.load(memory.hash(&3), &3).await.unwrap().kv().unwrap();
         assert_eq!(r3, (3, vec![3; 7 * KB]));
-        let r4v2 = store.load(memory.hash(&4)).await.unwrap().kv().unwrap();
+        let r4v2 = store.load(memory.hash(&4), &4).await.unwrap().kv().unwrap();
         assert_eq!(r4v2, (4, vec![!4; 3 * KB]));
-        let r5 = store.load(memory.hash(&5)).await.unwrap().kv().unwrap();
+        let r5 = store.load(memory.hash(&5), &5).await.unwrap().kv().unwrap();
         assert_eq!(r5, (5, vec![5; 11 * KB]));
-        let r6 = store.load(memory.hash(&6)).await.unwrap().kv().unwrap();
+        let r6 = store.load(memory.hash(&6), &6).await.unwrap().kv().unwrap();
         assert_eq!(r6, (6, vec![6; 7 * KB]));
 
         store.close().await.unwrap();
@@ -1067,15 +1073,15 @@ mod tests {
 
         let store = engine_for_test(dir.path()).await;
 
-        assert!(store.load(memory.hash(&1)).await.unwrap().kv().is_none());
-        assert!(store.load(memory.hash(&2)).await.unwrap().kv().is_none());
-        let r3 = store.load(memory.hash(&3)).await.unwrap().kv().unwrap();
+        assert!(store.load(memory.hash(&1), &1).await.unwrap().kv().is_none());
+        assert!(store.load(memory.hash(&2), &2).await.unwrap().kv().is_none());
+        let r3 = store.load(memory.hash(&3), &3).await.unwrap().kv().unwrap();
         assert_eq!(r3, (3, vec![3; 7 * KB]));
-        let r4v2 = store.load(memory.hash(&4)).await.unwrap().kv().unwrap();
+        let r4v2 = store.load(memory.hash(&4), &4).await.unwrap().kv().unwrap();
         assert_eq!(r4v2, (4, vec![!4; 3 * KB]));
-        let r5 = store.load(memory.hash(&5)).await.unwrap().kv().unwrap();
+        let r5 = store.load(memory.hash(&5), &5).await.unwrap().kv().unwrap();
         assert_eq!(r5, (5, vec![5; 11 * KB]));
-        let r6 = store.load(memory.hash(&6)).await.unwrap().kv().unwrap();
+        let r6 = store.load(memory.hash(&6), &6).await.unwrap().kv().unwrap();
         assert_eq!(r6, (6, vec![6; 7 * KB]));
     }
 
@@ -1096,14 +1102,14 @@ mod tests {
 
         for i in 0..9 {
             assert_eq!(
-                store.load(memory.hash(&i)).await.unwrap().kv(),
+                store.load(memory.hash(&i), &i).await.unwrap().kv(),
                 Some((i, vec![i as u8; 3 * KB]))
             );
         }
 
-        store.delete(memory.hash(&3));
+        store.delete(memory.hash(&3), &3);
         store.wait().await;
-        assert_eq!(store.load(memory.hash(&3)).await.unwrap().kv(), None);
+        assert_eq!(store.load(memory.hash(&3), &3).await.unwrap().kv(), None);
 
         store.close().await.unwrap();
         drop(store);
@@ -1112,18 +1118,18 @@ mod tests {
         for i in 0..9 {
             if i != 3 {
                 assert_eq!(
-                    store.load(memory.hash(&i)).await.unwrap().kv(),
+                    store.load(memory.hash(&i), &i).await.unwrap().kv(),
                     Some((i, vec![i as u8; 3 * KB]))
                 );
             } else {
-                assert_eq!(store.load(memory.hash(&3)).await.unwrap().kv(), None);
+                assert_eq!(store.load(memory.hash(&3), &3).await.unwrap().kv(), None);
             }
         }
 
         enqueue(&store, es[3].clone());
         store.wait().await;
         assert_eq!(
-            store.load(memory.hash(&3)).await.unwrap().kv(),
+            store.load(memory.hash(&3), &3).await.unwrap().kv(),
             Some((3, vec![3; 3 * KB]))
         );
 
@@ -1133,7 +1139,7 @@ mod tests {
         let store = store_for_test_with_tombstone_log(dir.path()).await;
 
         assert_eq!(
-            store.load(memory.hash(&3)).await.unwrap().kv(),
+            store.load(memory.hash(&3), &3).await.unwrap().kv(),
             Some((3, vec![3; 3 * KB]))
         );
     }
@@ -1157,14 +1163,14 @@ mod tests {
 
         for i in 0..9 {
             assert_eq!(
-                store.load(memory.hash(&i)).await.unwrap().kv(),
+                store.load(memory.hash(&i), &i).await.unwrap().kv(),
                 Some((i, vec![i as u8; 3 * KB]))
             );
         }
 
-        store.delete(memory.hash(&3));
+        store.delete(memory.hash(&3), &3);
         store.wait().await;
-        assert_eq!(store.load(memory.hash(&3)).await.unwrap().kv(), None);
+        assert_eq!(store.load(memory.hash(&3), &3).await.unwrap().kv(), None);
 
         store.destroy().await.unwrap();
 
@@ -1173,13 +1179,13 @@ mod tests {
 
         let store = store_for_test_with_tombstone_log(dir.path()).await;
         for i in 0..9 {
-            assert_eq!(store.load(memory.hash(&i)).await.unwrap().kv(), None);
+            assert_eq!(store.load(memory.hash(&i), &i).await.unwrap().kv(), None);
         }
 
         enqueue(&store, es[3].clone());
         store.wait().await;
         assert_eq!(
-            store.load(memory.hash(&3)).await.unwrap().kv(),
+            store.load(memory.hash(&3), &3).await.unwrap().kv(),
             Some((3, vec![3; 3 * KB]))
         );
 
@@ -1189,7 +1195,7 @@ mod tests {
         let store = store_for_test_with_tombstone_log(dir.path()).await;
 
         assert_eq!(
-            store.load(memory.hash(&3)).await.unwrap().kv(),
+            store.load(memory.hash(&3), &3).await.unwrap().kv(),
             Some((3, vec![3; 3 * KB]))
         );
     }
@@ -1234,7 +1240,7 @@ mod tests {
         }
 
         for i in 0..9 {
-            let r = store.load(memory.hash(&i)).await.unwrap().kv().unwrap();
+            let r = store.load(memory.hash(&i), &i).await.unwrap().kv().unwrap();
             assert_eq!(r, (i, vec![i as u8; 3 * KB]));
         }
 
@@ -1244,7 +1250,7 @@ mod tests {
         store.wait().await;
         let mut res = vec![];
         for i in 0..11 {
-            res.push(store.load(memory.hash(&i)).await.unwrap().kv());
+            res.push(store.load(memory.hash(&i), &i).await.unwrap().kv());
         }
         assert_eq!(
             res,
@@ -1268,7 +1274,7 @@ mod tests {
         store.wait().await;
         let mut res = vec![];
         for i in 0..12 {
-            res.push(store.load(memory.hash(&i)).await.unwrap().kv());
+            res.push(store.load(memory.hash(&i), &i).await.unwrap().kv());
         }
         assert_eq!(
             res,
@@ -1289,7 +1295,7 @@ mod tests {
         );
 
         // [[(11), (3), (5)], [(12), (13), (14)], [], [(9), (10), (1)]]
-        store.delete(memory.hash(&7));
+        store.delete(memory.hash(&7), &7);
         store.wait().await;
         enqueue(&store, es[12].clone());
         store.wait().await;
@@ -1299,7 +1305,7 @@ mod tests {
         store.wait().await;
         let mut res = vec![];
         for i in 0..15 {
-            res.push(store.load(memory.hash(&i)).await.unwrap().kv());
+            res.push(store.load(memory.hash(&i), &i).await.unwrap().kv());
         }
         assert_eq!(
             res,
@@ -1336,7 +1342,7 @@ mod tests {
         store.wait().await;
 
         // check entry 1
-        let r1 = store.load(memory.hash(&1)).await.unwrap().kv().unwrap();
+        let r1 = store.load(memory.hash(&1), &1).await.unwrap().kv().unwrap();
         assert_eq!(r1, (1, vec![1; 7 * KB]));
 
         // corrupt entry and header
@@ -1359,7 +1365,7 @@ mod tests {
             }
         }
 
-        assert!(store.load(memory.hash(&1)).await.unwrap().kv().is_none());
+        assert!(store.load(memory.hash(&1), &1).await.unwrap().kv().is_none());
     }
 
     #[test_log::test(tokio::test)]
@@ -1390,9 +1396,7 @@ mod tests {
             .with_device(d3)
             .build()
             .unwrap();
-        let engine = BlockEngineConfig::<u64, Vec<u8>, TestProperties>::new(device)
-            .with_block_size(64 * KB)
-            .boxed()
+        let engine = Box::new(BlockEngineConfig::<u64, Vec<u8>, TestProperties>::new(device).with_block_size(64 * KB))
             .build(EngineBuildContext {
                 io_engine,
                 metrics: Arc::new(Metrics::noop()),
diff --git a/foyer-storage/src/engine/mod.rs b/foyer-storage/src/engine/mod.rs
index e59bdbc8..8c902463 100644
--- a/foyer-storage/src/engine/mod.rs
+++ b/foyer-storage/src/engine/mod.rs
@@ -12,8 +12,9 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-use std::{any::Any, fmt::Debug, sync::Arc};
+use std::{any::Any, fmt::Debug, future::Future, hash::Hash, sync::Arc};
 
+use equivalent::Equivalent;
 use foyer_common::{
     code::{StorageKey, StorageValue},
     error::Result,
@@ -133,23 +134,17 @@ pub struct EngineBuildContext {
 }
 
 /// Disk cache engine config trait.
-#[expect(clippy::type_complexity)]
 pub trait EngineConfig<K, V, P>: Send + Sync + 'static + Debug
 where
     K: StorageKey,
     V: StorageValue,
     P: Properties,
 {
-    /// Build the engine with the given configurations.
-    fn build(self: Box<Self>, ctx: EngineBuildContext) -> BoxFuture<'static, Result<Arc<dyn Engine<K, V, P>>>>;
+    /// The concrete engine type produced by this config.
+    type Engine: Engine<K, V, P>;
 
-    /// Box the config.
-    fn boxed(self) -> Box<Self>
-    where
-        Self: Sized,
-    {
-        Box::new(self)
-    }
+    /// Build the engine with the given configurations.
+    fn build(self: Box<Self>, ctx: EngineBuildContext) -> BoxFuture<'static, Result<Arc<Self::Engine>>>;
 }
 
 /// Disk cache engine trait.
@@ -159,6 +154,9 @@ where
     V: StorageValue,
     P: Properties,
 {
+    /// Whether this is a noop (disabled) engine.
+    const IS_NOOP: bool = false;
+
     /// Get the device used by this disk cache engine.
     fn device(&self) -> &Arc<dyn Device>;
 
@@ -169,29 +167,32 @@ where
     fn enqueue(&self, piece: PieceRef<K, V, P>, estimated_size: usize);
 
     /// Load a cache entry from the disk cache.
-    ///
-    /// `load` may return a false-positive result on entry key hash collision. It's the caller's responsibility to
-    /// check if the returned key matches the given key.
-    fn load(&self, hash: u64) -> BoxFuture<'static, Result<Load<K, V, P>>>;
+    fn load<'a, Q>(&'a self, hash: u64, key: &'a Q) -> impl Future<Output = Result<Load<K, V, P>>> + Send + 'a
+    where
+        Q: Hash + Equivalent<K> + Sync + ?Sized;
 
     /// Delete the cache entry with the given key from the disk cache.
-    fn delete(&self, hash: u64);
+    fn delete<Q>(&self, hash: u64, key: &Q)
+    where
+        Q: Hash + Equivalent<K> + Sync + ?Sized;
 
     /// Check if the disk cache contains a cached entry with the given key.
     ///
     /// `contains` may return a false-positive result if there is a hash collision with the given key.
-    fn may_contains(&self, hash: u64) -> bool;
+    fn may_contains<Q>(&self, hash: u64, key: &Q) -> bool
+    where
+        Q: Hash + Equivalent<K> + Sync + ?Sized;
 
     /// Delete all cached entries of the disk cache.
-    fn destroy(&self) -> BoxFuture<'static, Result<()>>;
+    fn destroy(&self) -> impl Future<Output = Result<()>> + Send + '_;
 
     /// Wait for the ongoing flush and reclaim tasks to finish.
-    fn wait(&self) -> BoxFuture<'static, ()>;
+    fn wait(&self) -> impl Future<Output = ()> + Send + '_;
 
     /// Close the disk cache gracefully.
     ///
     /// `close` will wait for all ongoing flush and reclaim tasks to finish.
-    fn close(&self) -> BoxFuture<'static, Result<()>>;
+    fn close(&self) -> impl Future<Output = Result<()>> + Send + '_;
 }
 
 pub mod block;
diff --git a/foyer-storage/src/engine/noop.rs b/foyer-storage/src/engine/noop.rs
index f0a65cc8..b9251936 100644
--- a/foyer-storage/src/engine/noop.rs
+++ b/foyer-storage/src/engine/noop.rs
@@ -12,8 +12,9 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-use std::{fmt::Debug, marker::PhantomData, sync::Arc};
+use std::{fmt::Debug, hash::Hash, marker::PhantomData, sync::Arc};
 
+use equivalent::Equivalent;
 use foyer_common::{
     code::{StorageKey, StorageValue},
     error::Result,
@@ -28,6 +29,7 @@ use crate::{
     keeper::PieceRef,
 };
 
+/// Config for the noop disk cache engine that does nothing.
 pub struct NoopEngineConfig<K, V, P>(PhantomData<(K, V, P)>)
 where
     K: StorageKey,
@@ -62,6 +64,7 @@ where
     V: StorageValue,
     P: Properties,
 {
+    /// Build the noop engine.
     pub fn build(self) -> Arc<NoopEngine<K, V, P>> {
         let device = NoopDeviceBuilder::default().build().unwrap();
         let device: Arc<dyn Device> = device;
@@ -78,22 +81,14 @@ where
     V: StorageValue,
     P: Properties,
 {
-    fn build(self: Box<Self>, _: EngineBuildContext) -> BoxFuture<'static, Result<Arc<dyn Engine<K, V, P>>>> {
-        async move { Ok((*self).build() as Arc<dyn Engine<K, V, P>>) }.boxed()
-    }
-}
+    type Engine = NoopEngine<K, V, P>;
 
-impl<K, V, P> From<NoopEngineConfig<K, V, P>> for Box<dyn EngineConfig<K, V, P>>
-where
-    K: StorageKey,
-    V: StorageValue,
-    P: Properties,
-{
-    fn from(builder: NoopEngineConfig<K, V, P>) -> Self {
-        builder.boxed()
+    fn build(self: Box<Self>, _: EngineBuildContext) -> BoxFuture<'static, Result<Arc<NoopEngine<K, V, P>>>> {
+        async move { Ok((*self).build()) }.boxed()
     }
 }
 
+/// Noop disk cache engine that does nothing.
 pub struct NoopEngine<K, V, P>
 where
     K: StorageKey,
@@ -121,6 +116,8 @@ where
     V: StorageValue,
     P: Properties,
 {
+    const IS_NOOP: bool = true;
+
     fn device(&self) -> &Arc<dyn Device> {
         &self.device
     }
@@ -131,25 +128,33 @@ where
 
     fn enqueue(&self, _: PieceRef<K, V, P>, _: usize) {}
 
-    fn load(&self, _: u64) -> BoxFuture<'static, Result<Load<K, V, P>>> {
-        async move { Ok(Load::Miss) }.boxed()
+    async fn load<'a, Q>(&'a self, _: u64, _: &'a Q) -> Result<Load<K, V, P>>
+    where
+        Q: Hash + Equivalent<K> + Sync + ?Sized,
+    {
+        Ok(Load::Miss)
     }
 
-    fn delete(&self, _: u64) {}
+    fn delete<Q>(&self, _: u64, _: &Q)
+    where
+        Q: Hash + Equivalent<K> + Sync + ?Sized,
+    {
+    }
 
-    fn may_contains(&self, _: u64) -> bool {
+    fn may_contains<Q>(&self, _: u64, _: &Q) -> bool
+    where
+        Q: Hash + Equivalent<K> + Sync + ?Sized,
+    {
         false
     }
 
-    fn destroy(&self) -> BoxFuture<'static, Result<()>> {
-        async move { Ok(()) }.boxed()
+    async fn destroy(&self) -> Result<()> {
+        Ok(())
     }
 
-    fn wait(&self) -> BoxFuture<'static, ()> {
-        async move {}.boxed()
-    }
+    async fn wait(&self) {}
 
-    fn close(&self) -> BoxFuture<'static, Result<()>> {
-        async move { Ok(()) }.boxed()
+    async fn close(&self) -> Result<()> {
+        Ok(())
     }
 }
diff --git a/foyer-storage/src/keeper.rs b/foyer-storage/src/keeper.rs
index cdbf8c36..5d728e30 100644
--- a/foyer-storage/src/keeper.rs
+++ b/foyer-storage/src/keeper.rs
@@ -93,6 +93,7 @@ where
     }
 }
 
+/// A reference to a piece in the keeper, removing it from the keeper on drop.
 pub struct PieceRef<K, V, P>
 where
     K: StorageKey,
diff --git a/foyer-storage/src/prelude.rs b/foyer-storage/src/prelude.rs
index 3b98e1e6..3e43d407 100644
--- a/foyer-storage/src/prelude.rs
+++ b/foyer-storage/src/prelude.rs
@@ -19,10 +19,11 @@ pub use crate::{
     engine::{
         Engine, EngineBuildContext, EngineConfig, Load, Populated, RecoverMode,
         block::{
-            engine::BlockEngineConfig,
+            engine::{BlockEngine, BlockEngineConfig},
             eviction::{EvictionInfo, EvictionPicker, FifoPicker, InvalidRatioPicker},
             manager::{Block, BlockStatistics},
         },
+        noop::{NoopEngine, NoopEngineConfig},
     },
     filter::{
         StorageFilter, StorageFilterCondition, StorageFilterResult,
@@ -45,5 +46,6 @@ pub use crate::{
             psync::{PsyncIoEngine, PsyncIoEngineConfig},
         },
     },
+    keeper::PieceRef,
     store::{Store, StoreBuilder},
 };
diff --git a/foyer-storage/src/store.rs b/foyer-storage/src/store.rs
index 73e1b91d..11066855 100644
--- a/foyer-storage/src/store.rs
+++ b/foyer-storage/src/store.rs
@@ -12,14 +12,7 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-use std::{
-    any::{Any, TypeId},
-    borrow::Cow,
-    fmt::Debug,
-    hash::Hash,
-    sync::Arc,
-    time::Instant,
-};
+use std::{borrow::Cow, fmt::Debug, hash::Hash, sync::Arc, time::Instant};
 
 use equivalent::Equivalent;
 use foyer_common::{
@@ -36,10 +29,7 @@ use crate::test_utils::*;
 use crate::{
     StorageFilterResult,
     compress::Compression,
-    engine::{
-        Engine, EngineBuildContext, EngineConfig, Load, Populated, RecoverMode,
-        noop::{NoopEngine, NoopEngineConfig},
-    },
+    engine::{Engine, EngineBuildContext, EngineConfig, Load, Populated, RecoverMode, noop::NoopEngineConfig},
     io::{
         device::{Device, statistics::Statistics, throttle::Throttle},
         engine::{IoEngineBuildContext, IoEngineConfig, monitor::MonitoredIoEngine, psync::PsyncIoEngineConfig},
@@ -49,27 +39,29 @@ use crate::{
 };
 
 /// The disk cache engine that serves as the storage backend of `foyer`.
-pub struct Store<K, V, S, P>
+pub struct Store<K, V, S, P, E>
 where
     K: StorageKey,
     V: StorageValue,
     S: HashBuilder + Debug,
     P: Properties,
+    E: Engine<K, V, P>,
 {
-    inner: Arc<StoreInner<K, V, S, P>>,
+    inner: Arc<StoreInner<K, V, S, P, E>>,
 }
 
-struct StoreInner<K, V, S, P>
+struct StoreInner<K, V, S, P, E>
 where
     K: StorageKey,
     V: StorageValue,
     S: HashBuilder + Debug,
     P: Properties,
+    E: Engine<K, V, P>,
 {
     hasher: Arc<S>,
 
     keeper: Keeper<K, V, P>,
-    engine: Arc<dyn Engine<K, V, P>>,
+    engine: Arc<E>,
 
     compression: Compression,
 
@@ -81,12 +73,13 @@ where
     load_throttle_switch: LoadThrottleSwitch,
 }
 
-impl<K, V, S, P> Debug for Store<K, V, S, P>
+impl<K, V, S, P, E> Debug for Store<K, V, S, P, E>
 where
     K: StorageKey,
     V: StorageValue,
     S: HashBuilder + Debug,
     P: Properties,
+    E: Engine<K, V, P>,
 {
     fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
         f.debug_struct("Store")
@@ -98,12 +91,13 @@ where
     }
 }
 
-impl<K, V, S, P> Clone for Store<K, V, S, P>
+impl<K, V, S, P, E> Clone for Store<K, V, S, P, E>
 where
     K: StorageKey,
     V: StorageValue,
     S: HashBuilder + Debug,
     P: Properties,
+    E: Engine<K, V, P>,
 {
     fn clone(&self) -> Self {
         Self {
@@ -112,12 +106,13 @@ where
     }
 }
 
-impl<K, V, S, P> Store<K, V, S, P>
+impl<K, V, S, P, E> Store<K, V, S, P, E>
 where
     K: StorageKey,
     V: StorageValue,
     S: HashBuilder + Debug,
     P: Properties,
+    E: Engine<K, V, P>,
 {
     /// Close the disk cache gracefully.
     ///
@@ -161,7 +156,7 @@ where
     /// Load a cache entry from the disk cache.
     pub async fn load<Q>(&self, key: &Q) -> Result<Load<K, V, P>>
     where
-        Q: Hash + Equivalent<K> + ?Sized,
+        Q: Hash + Equivalent<K> + Sync + ?Sized,
     {
         let now = Instant::now();
 
@@ -185,39 +180,14 @@ where
             return Ok(Load::Throttled);
         }
 
-        match self.inner.engine.load(hash).await {
-            Ok(Load::Entry {
-                key: k,
-                value: v,
-                populated: p,
-            }) if key.equivalent(&k) => {
-                self.inner.metrics.storage_hit.increase(1);
-                self.inner
-                    .metrics
-                    .storage_hit_duration
-                    .record(now.elapsed().as_secs_f64());
-                Ok(Load::Entry {
-                    key: k,
-                    value: v,
-                    populated: p,
-                })
-            }
-            Ok(Load::Piece { piece, populated }) if key.equivalent(piece.key()) => {
+        match self.inner.engine.load(hash, key).await {
+            Ok(load @ (Load::Entry { .. } | Load::Piece { .. })) => {
                 self.inner.metrics.storage_hit.increase(1);
                 self.inner
                     .metrics
                     .storage_hit_duration
                     .record(now.elapsed().as_secs_f64());
-                Ok(Load::Piece { piece, populated })
-            }
-            Ok(Load::Entry { .. }) | Ok(Load::Piece { .. }) => {
-                self.inner.metrics.storage_miss.increase(1);
-                self.inner.metrics.storage_false_positive.increase(1);
-                self.inner
-                    .metrics
-                    .storage_miss_duration
-                    .record(now.elapsed().as_secs_f64());
-                Ok(Load::Miss)
+                Ok(load)
             }
             Ok(Load::Miss) => {
                 self.inner.metrics.storage_miss.increase(1);
@@ -245,12 +215,12 @@ where
     /// Delete the cache entry with the given key from the disk cache.
     pub fn delete<'a, Q>(&'a self, key: &'a Q)
     where
-        Q: Hash + Equivalent<K> + ?Sized,
+        Q: Hash + Equivalent<K> + Sync + ?Sized,
     {
         let now = Instant::now();
 
         let hash = self.inner.hasher.hash_one(key);
-        self.inner.engine.delete(hash);
+        self.inner.engine.delete(hash, key);
 
         self.inner.metrics.storage_delete.increase(1);
         self.inner
@@ -264,10 +234,10 @@ where
     /// `contains` may return a false-positive result if there is a hash collision with the given key.
     pub fn may_contains<Q>(&self, key: &Q) -> bool
     where
-        Q: Hash + Equivalent<K> + ?Sized,
+        Q: Hash + Equivalent<K> + Sync + ?Sized,
     {
         let hash = self.inner.hasher.hash_one(key);
-        self.inner.engine.may_contains(hash)
+        self.inner.engine.may_contains(hash, key)
     }
 
     /// Delete all cached entries of the disk cache.
@@ -313,24 +283,25 @@ where
 
     /// If the disk cache is enabled.
     pub fn is_enabled(&self) -> bool {
-        self.inner.engine.type_id() != TypeId::of::<Arc<NoopEngine<K, V, P>>>()
+        !E::IS_NOOP
     }
 }
 
 /// The builder of the disk cache.
-pub struct StoreBuilder<K, V, S, P>
+pub struct StoreBuilder<K, V, S, P, EC = NoopEngineConfig<K, V, P>>
 where
     K: StorageKey,
     V: StorageValue,
     S: HashBuilder + Debug,
     P: Properties,
+    EC: EngineConfig<K, V, P>,
 {
     name: Cow<'static, str>,
     memory: Cache<K, V, S, P>,
     metrics: Arc<Metrics>,
 
     io_engine_config: Option<Box<dyn IoEngineConfig>>,
-    engine_config: Option<Box<dyn EngineConfig<K, V, P>>>,
+    engine_config: EC,
 
     spawner: Option<Spawner>,
 
@@ -341,12 +312,13 @@ where
     load_throttle_switch: LoadThrottleSwitch,
 }
 
-impl<K, V, S, P> Debug for StoreBuilder<K, V, S, P>
+impl<K, V, S, P, EC> Debug for StoreBuilder<K, V, S, P, EC>
 where
     K: StorageKey,
     V: StorageValue,
     S: HashBuilder + Debug,
     P: Properties,
+    EC: EngineConfig<K, V, P>,
 {
     fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
         f.debug_struct("StoreBuilder")
@@ -362,7 +334,7 @@ where
     }
 }
 
-impl<K, V, S, P> StoreBuilder<K, V, S, P>
+impl<K, V, S, P> StoreBuilder<K, V, S, P, NoopEngineConfig<K, V, P>>
 where
     K: StorageKey,
     V: StorageValue,
@@ -377,7 +349,7 @@ where
             metrics,
 
             io_engine_config: None,
-            engine_config: None,
+            engine_config: NoopEngineConfig::default(),
 
             spawner: None,
 
@@ -387,7 +359,16 @@ where
             load_throttle_switch: LoadThrottleSwitch::default(),
         }
     }
+}
 
+impl<K, V, S, P, EC> StoreBuilder<K, V, S, P, EC>
+where
+    K: StorageKey,
+    V: StorageValue,
+    S: HashBuilder + Debug,
+    P: Properties,
+    EC: EngineConfig<K, V, P>,
+{
     /// Set io engine config for the disk cache store.
     ///
     /// Default: [`crate::io::engine::psync::PsyncIoEngineConfig`].
@@ -397,9 +378,19 @@ where
     }
 
     /// Set engine config for the disk cache store.
-    pub fn with_engine_config(mut self, config: impl Into<Box<dyn EngineConfig<K, V, P>>>) -> Self {
-        self.engine_config = Some(config.into());
-        self
+    pub fn with_engine_config<EC2: EngineConfig<K, V, P>>(self, config: EC2) -> StoreBuilder<K, V, S, P, EC2> {
+        StoreBuilder {
+            name: self.name,
+            memory: self.memory,
+            metrics: self.metrics,
+            io_engine_config: self.io_engine_config,
+            engine_config: config,
+            spawner: self.spawner,
+            compression: self.compression,
+            recover_mode: self.recover_mode,
+            #[cfg(any(test, feature = "test_utils"))]
+            load_throttle_switch: self.load_throttle_switch,
+        }
     }
 
     /// Set the compression algorithm of the disk cache store.
@@ -441,11 +432,11 @@ where
 
     #[doc(hidden)]
     pub fn is_noop(&self) -> bool {
-        self.engine_config.is_none()
+        EC::Engine::IS_NOOP
     }
 
     /// Build the disk cache store with the given configuration.
-    pub async fn build(self) -> Result<Store<K, V, S, P>> {
+    pub async fn build(self) -> Result<Store<K, V, S, P, EC::Engine>> {
         let memory = self.memory;
         let metrics = self.metrics;
 
@@ -469,18 +460,13 @@ where
             .await?;
         let io_engine = MonitoredIoEngine::new(io_engine, metrics.clone());
 
-        let engine_builder = match self.engine_config {
-            Some(eb) => eb,
-            None => {
-                tracing::info!(
-                    "[store builder]: No engine builder is provided, run disk cache in mock mode that do nothing."
-                );
-
-                Box::<NoopEngineConfig<K, V, P>>::default()
-            }
-        };
+        if EC::Engine::IS_NOOP {
+            tracing::info!(
+                "[store builder]: No-Op engine builder is provided, run disk cache in mock mode that do nothing."
+            );
+        }
 
-        let engine = engine_builder
+        let engine = Box::new(self.engine_config)
             .build(EngineBuildContext {
                 io_engine,
                 metrics: metrics.clone(),
diff --git a/foyer-storage/tests/storage_fuzzy_test.rs b/foyer-storage/tests/storage_fuzzy_test.rs
index ec792814..ff3296b9 100644
--- a/foyer-storage/tests/storage_fuzzy_test.rs
+++ b/foyer-storage/tests/storage_fuzzy_test.rs
@@ -34,7 +34,13 @@ async fn test_store(
     memory: Cache<u64, Vec<u8>, ModHasher, TestProperties>,
     builder: impl Fn(
         &Cache<u64, Vec<u8>, ModHasher, TestProperties>,
-    ) -> StoreBuilder<u64, Vec<u8>, ModHasher, TestProperties>,
+    ) -> StoreBuilder<
+        u64,
+        Vec<u8>,
+        ModHasher,
+        TestProperties,
+        BlockEngineConfig<u64, Vec<u8>, TestProperties>,
+    >,
     recorder: Recorder,
 ) {
     let store = builder(&memory).build().await.unwrap();
@@ -105,7 +111,7 @@ fn basic(
     memory: &Cache<u64, Vec<u8>, ModHasher, TestProperties>,
     path: impl AsRef<Path>,
     recorder: &Recorder,
-) -> StoreBuilder<u64, Vec<u8>, ModHasher, TestProperties> {
+) -> StoreBuilder<u64, Vec<u8>, ModHasher, TestProperties, BlockEngineConfig<u64, Vec<u8>, TestProperties>> {
     // TODO(MrCroxx): Test mixed engine here.
     StoreBuilder::new("test", memory.clone(), Arc::new(Metrics::noop())).with_engine_config(
         BlockEngineConfig::new(FsDeviceBuilder::new(path).with_capacity(4 * MB).build().unwrap())
diff --git a/foyer/src/hybrid/builder.rs b/foyer/src/hybrid/builder.rs
index d6ceea2c..610de1cd 100644
--- a/foyer/src/hybrid/builder.rs
+++ b/foyer/src/hybrid/builder.rs
@@ -24,7 +24,7 @@ use foyer_common::{
     spawn::Spawner,
 };
 use foyer_memory::{Cache, CacheBuilder, EvictionConfig, Filter, Weighter};
-use foyer_storage::{Compression, EngineConfig, IoEngineConfig, RecoverMode, StoreBuilder};
+use foyer_storage::{Compression, Engine, EngineConfig, IoEngineConfig, NoopEngineConfig, RecoverMode, StoreBuilder};
 use mixtrics::{metrics::BoxedRegistry, registry::noop::NoopMetricsRegistry};
 
 use crate::hybrid::cache::{
@@ -237,24 +237,26 @@ where
 }
 
 /// Hybrid cache builder modify the disk cache configurations.
-pub struct HybridCacheBuilderPhaseStorage<K, V, S>
+pub struct HybridCacheBuilderPhaseStorage<K, V, S, EC = NoopEngineConfig<K, V, HybridCacheProperties>>
 where
     K: StorageKey,
     V: StorageValue,
     S: HashBuilder + Debug,
+    EC: EngineConfig<K, V, HybridCacheProperties>,
 {
     name: Cow<'static, str>,
     options: HybridCacheOptions,
     metrics: Arc<Metrics>,
     memory: Cache<K, V, S, HybridCacheProperties>,
-    builder: StoreBuilder<K, V, S, HybridCacheProperties>,
+    builder: StoreBuilder<K, V, S, HybridCacheProperties, EC>,
 }
 
-impl<K, V, S> HybridCacheBuilderPhaseStorage<K, V, S>
+impl<K, V, S, EC> HybridCacheBuilderPhaseStorage<K, V, S, EC>
 where
     K: StorageKey,
     V: StorageValue,
     S: HashBuilder + Debug,
+    EC: EngineConfig<K, V, HybridCacheProperties>,
 {
     /// Set io engine config for the disk cache store.
     pub fn with_io_engine_config(self, config: impl Into<Box<dyn IoEngineConfig>>) -> Self {
@@ -269,9 +271,12 @@ where
     }
 
     /// Set engine config for the disk cache store.
-    pub fn with_engine_config(self, config: impl Into<Box<dyn EngineConfig<K, V, HybridCacheProperties>>>) -> Self {
+    pub fn with_engine_config<EC2: EngineConfig<K, V, HybridCacheProperties>>(
+        self,
+        config: EC2,
+    ) -> HybridCacheBuilderPhaseStorage<K, V, S, EC2> {
         let builder = self.builder.with_engine_config(config);
-        Self {
+        HybridCacheBuilderPhaseStorage {
             name: self.name,
             options: self.options,
             metrics: self.metrics,
@@ -329,7 +334,10 @@ where
     }
 
     /// Build and open the hybrid cache with the given configurations.
-    pub async fn build(self) -> Result<HybridCache<K, V, S>> {
+    pub async fn build(self) -> Result<HybridCache<K, V, S, EC::Engine>>
+    where
+        EC::Engine: Engine<K, V, HybridCacheProperties>,
+    {
         let builder = self.builder;
 
         let piped = match (builder.is_noop(), self.options.policy) {
diff --git a/foyer/src/hybrid/cache.rs b/foyer/src/hybrid/cache.rs
index 179947f8..ac9ee486 100644
--- a/foyer/src/hybrid/cache.rs
+++ b/foyer/src/hybrid/cache.rs
@@ -41,7 +41,7 @@ use foyer_common::{
     rate::RateLimiter,
 };
 use foyer_memory::{Cache, CacheEntry, FetchTarget, GetOrFetch, Piece, Pipe};
-use foyer_storage::{Load, Populated, Statistics, Store};
+use foyer_storage::{BlockEngine, Engine, Load, Populated, Statistics, Store};
 use futures_util::FutureExt as _;
 use pin_project::pin_project;
 use serde::{Deserialize, Serialize};
@@ -183,42 +183,46 @@ pub enum HybridCachePolicy {
     WriteOnInsertion,
 }
 
-pub struct HybridCachePipe<K, V, S>
+pub struct HybridCachePipe<K, V, S, E = BlockEngine<K, V, HybridCacheProperties>>
 where
     K: StorageKey,
     V: StorageValue,
     S: HashBuilder + Debug,
+    E: Engine<K, V, HybridCacheProperties>,
 {
-    store: Store<K, V, S, HybridCacheProperties>,
+    store: Store<K, V, S, HybridCacheProperties, E>,
 }
 
-impl<K, V, S> Debug for HybridCachePipe<K, V, S>
+impl<K, V, S, E> Debug for HybridCachePipe<K, V, S, E>
 where
     K: StorageKey,
     V: StorageValue,
     S: HashBuilder + Debug,
+    E: Engine<K, V, HybridCacheProperties>,
 {
     fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
         f.debug_struct("HybridCachePipe").finish()
     }
 }
 
-impl<K, V, S> HybridCachePipe<K, V, S>
+impl<K, V, S, E> HybridCachePipe<K, V, S, E>
 where
     K: StorageKey,
     V: StorageValue,
     S: HashBuilder + Debug,
+    E: Engine<K, V, HybridCacheProperties>,
 {
-    pub fn new(store: Store<K, V, S, HybridCacheProperties>) -> Self {
+    pub fn new(store: Store<K, V, S, HybridCacheProperties, E>) -> Self {
         Self { store }
     }
 }
 
-impl<K, V, S> Pipe for HybridCachePipe<K, V, S>
+impl<K, V, S, E> Pipe for HybridCachePipe<K, V, S, E>
 where
     K: StorageKey,
     V: StorageValue,
     S: HashBuilder + Debug,
+    E: Engine<K, V, HybridCacheProperties>,
 {
     type Key = K;
     type Value = V;
@@ -285,11 +289,12 @@ impl Default for HybridCacheOptions {
     }
 }
 
-struct Inner<K, V, S = DefaultHasher>
+struct Inner<K, V, S = DefaultHasher, E = BlockEngine<K, V, HybridCacheProperties>>
 where
     K: StorageKey,
     V: StorageValue,
     S: HashBuilder + Debug,
+    E: Engine<K, V, HybridCacheProperties>,
 {
     name: Cow<'static, str>,
     policy: HybridCachePolicy,
@@ -297,23 +302,24 @@ where
     metrics: Arc<Metrics>,
     closed: Arc<AtomicBool>,
     memory: Cache<K, V, S, HybridCacheProperties>,
-    storage: Store<K, V, S, HybridCacheProperties>,
+    storage: Store<K, V, S, HybridCacheProperties, E>,
     #[cfg(feature = "tracing")]
     tracing: std::sync::atomic::AtomicBool,
     #[cfg(feature = "tracing")]
     tracing_config: TracingConfig,
 }
 
-impl<K, V, S> Inner<K, V, S>
+impl<K, V, S, E> Inner<K, V, S, E>
 where
     K: StorageKey,
     V: StorageValue,
     S: HashBuilder + Debug,
+    E: Engine<K, V, HybridCacheProperties>,
 {
     async fn close_inner(
         closed: Arc<AtomicBool>,
         memory: Cache<K, V, S, HybridCacheProperties>,
-        storage: Store<K, V, S, HybridCacheProperties>,
+        storage: Store<K, V, S, HybridCacheProperties, E>,
         flush_on_close: bool,
     ) -> Result<()> {
         if closed.fetch_or(true, Ordering::Relaxed) {
@@ -344,11 +350,12 @@ where
     }
 }
 
-impl<K, V, S> Drop for Inner<K, V, S>
+impl<K, V, S, E> Drop for Inner<K, V, S, E>
 where
     K: StorageKey,
     V: StorageValue,
     S: HashBuilder + Debug,
+    E: Engine<K, V, HybridCacheProperties>,
 {
     fn drop(&mut self) {
         let name = self.name.clone();
@@ -378,20 +385,22 @@ where
 /// ```
 ///
 /// So, `storage` must not hold the reference of `memory`.
-pub struct HybridCache<K, V, S = DefaultHasher>
+pub struct HybridCache<K, V, S = DefaultHasher, E = BlockEngine<K, V, HybridCacheProperties>>
 where
     K: StorageKey,
     V: StorageValue,
     S: HashBuilder + Debug,
+    E: Engine<K, V, HybridCacheProperties>,
 {
-    inner: Arc<Inner<K, V, S>>,
+    inner: Arc<Inner<K, V, S, E>>,
 }
 
-impl<K, V, S> Debug for HybridCache<K, V, S>
+impl<K, V, S, E> Debug for HybridCache<K, V, S, E>
 where
     K: StorageKey,
     V: StorageValue,
     S: HashBuilder + Debug,
+    E: Engine<K, V, HybridCacheProperties>,
 {
     fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
         let mut r = f.debug_struct("HybridCache");
@@ -406,11 +415,12 @@ where
     }
 }
 
-impl<K, V, S> Clone for HybridCache<K, V, S>
+impl<K, V, S, E> Clone for HybridCache<K, V, S, E>
 where
     K: StorageKey,
     V: StorageValue,
     S: HashBuilder + Debug,
+    E: Engine<K, V, HybridCacheProperties>,
 {
     fn clone(&self) -> Self {
         Self {
@@ -419,7 +429,7 @@ where
     }
 }
 
-impl<K, V> HybridCache<K, V, DefaultHasher>
+impl<K, V> HybridCache<K, V, DefaultHasher, BlockEngine<K, V, HybridCacheProperties>>
 where
     K: StorageKey,
     V: StorageValue,
@@ -430,17 +440,18 @@ where
     }
 }
 
-impl<K, V, S> HybridCache<K, V, S>
+impl<K, V, S, E> HybridCache<K, V, S, E>
 where
     K: StorageKey,
     V: StorageValue,
     S: HashBuilder + Debug,
+    E: Engine<K, V, HybridCacheProperties>,
 {
     pub(crate) fn new(
         name: Cow<'static, str>,
         options: HybridCacheOptions,
         memory: Cache<K, V, S, HybridCacheProperties>,
-        storage: Store<K, V, S, HybridCacheProperties>,
+        storage: Store<K, V, S, HybridCacheProperties, E>,
         metrics: Arc<Metrics>,
     ) -> Self {
         let policy = options.policy;
@@ -493,7 +504,7 @@ where
     }
 
     /// Access the disk cache.
-    pub fn storage(&self) -> &Store<K, V, S, HybridCacheProperties> {
+    pub fn storage(&self) -> &Store<K, V, S, HybridCacheProperties, E> {
         &self.inner.storage
     }
 
@@ -600,7 +611,7 @@ where
     /// `contains` may return a false-positive result if there is a hash collision with the given key.
     pub fn contains<Q>(&self, key: &Q) -> bool
     where
-        Q: Hash + Equivalent<K> + ?Sized,
+        Q: Hash + Equivalent<K> + Sync + ?Sized,
     {
         self.inner.memory.contains(key) || self.inner.storage.may_contains(key)
     }
@@ -631,12 +642,12 @@ where
     }
 
     /// Create a new [`HybridCacheWriter`].
-    pub fn writer(&self, key: K) -> HybridCacheWriter<K, V, S> {
+    pub fn writer(&self, key: K) -> HybridCacheWriter<K, V, S, E> {
         HybridCacheWriter::new(self.clone(), key)
     }
 
     /// Create a new [`HybridCacheStorageWriter`].
-    pub fn storage_writer(&self, key: K) -> HybridCacheStorageWriter<K, V, S> {
+    pub fn storage_writer(&self, key: K) -> HybridCacheStorageWriter<K, V, S, E> {
         HybridCacheStorageWriter::new(self.clone(), key)
     }
 
@@ -651,11 +662,12 @@ where
     }
 }
 
-impl<K, V, S> HybridCache<K, V, S>
+impl<K, V, S, E> HybridCache<K, V, S, E>
 where
     K: StorageKey + Clone,
     V: StorageValue,
     S: HashBuilder + Debug,
+    E: Engine<K, V, HybridCacheProperties>,
 {
     /// Get cached entry with the given key from the hybrid cache.
     pub fn get<Q>(&self, key: &Q) -> HybridGet<K, V, S>
@@ -718,7 +730,7 @@ where
     }
 
     /// Get cached entry with the given key from the hybrid cache.
-    pub fn get_or_fetch<Q, F, FU, IT, ER>(&self, key: &Q, fetch: F) -> HybridGetOrFetch<K, V, S>
+    pub fn get_or_fetch<Q, F, FU, IT, ER>(&self, key: &Q, fetch: F) -> HybridGetOrFetch<K, V, S, E>
     where
         Q: Hash + Equivalent<K> + ?Sized + ToOwned<Owned = K>,
         F: FnOnce() -> FU,
@@ -922,16 +934,17 @@ struct GetOrFetchCtx {
 /// Future for [`HybridCache::get_or_fetch`].
 #[must_use]
 #[pin_project]
-pub struct HybridGetOrFetch<K, V, S = DefaultHasher>
+pub struct HybridGetOrFetch<K, V, S = DefaultHasher, E = BlockEngine<K, V, HybridCacheProperties>>
 where
     K: StorageKey,
     V: StorageValue,
     S: HashBuilder + Debug,
+    E: Engine<K, V, HybridCacheProperties>,
 {
     #[pin]
     inner: GetOrFetch<K, V, S, HybridCacheProperties>,
     policy: HybridCachePolicy,
-    store: Store<K, V, S, HybridCacheProperties>,
+    store: Store<K, V, S, HybridCacheProperties, E>,
     ctx: Arc<GetOrFetchCtx>,
     metrics: Arc<Metrics>,
     start: Instant,
@@ -941,22 +954,24 @@ where
     span_cancel_threshold: Duration,
 }
 
-impl<K, V, S> Debug for HybridGetOrFetch<K, V, S>
+impl<K, V, S, E> Debug for HybridGetOrFetch<K, V, S, E>
 where
     K: StorageKey,
     V: StorageValue,
     S: HashBuilder + Debug,
+    E: Engine<K, V, HybridCacheProperties>,
 {
     fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
         f.debug_struct("HybridGetOrFetch").field("inner", &self.inner).finish()
     }
 }
 
-impl<K, V, S> Future for HybridGetOrFetch<K, V, S>
+impl<K, V, S, E> Future for HybridGetOrFetch<K, V, S, E>
 where
     K: StorageKey + Clone,
     V: StorageValue,
     S: HashBuilder + Debug,
+    E: Engine<K, V, HybridCacheProperties>,
 {
     type Output = Result<HybridCacheEntry<K, V, S>>;
 
@@ -1006,11 +1021,12 @@ where
     }
 }
 
-impl<K, V, S> HybridGetOrFetch<K, V, S>
+impl<K, V, S, E> HybridGetOrFetch<K, V, S, E>
 where
     K: StorageKey + Clone,
     V: StorageValue,
     S: HashBuilder + Debug,
+    E: Engine<K, V, HybridCacheProperties>,
 {
     /// Check if the future need to be awaited or can be unwrap at once.
     pub fn need_await(&self) -> bool {
@@ -1051,7 +1067,9 @@ mod tests {
     const KB: usize = 1024;
     const MB: usize = 1024 * 1024;
 
-    async fn open(dir: impl AsRef<Path>) -> HybridCache<u64, Vec<u8>, ModHasher> {
+    async fn open(
+        dir: impl AsRef<Path>,
+    ) -> HybridCache<u64, Vec<u8>, ModHasher, BlockEngine<u64, Vec<u8>, HybridCacheProperties>> {
         open_with(dir, |b| b, |b| b).await
     }
 
@@ -1061,7 +1079,7 @@ mod tests {
         block_engine_builder_mapper: impl FnOnce(
             BlockEngineConfig<u64, Vec<u8>, HybridCacheProperties>,
         ) -> BlockEngineConfig<u64, Vec<u8>, HybridCacheProperties>,
-    ) -> HybridCache<u64, Vec<u8>, ModHasher> {
+    ) -> HybridCache<u64, Vec<u8>, ModHasher, BlockEngine<u64, Vec<u8>, HybridCacheProperties>> {
         let mut block_engine_builder =
             BlockEngineConfig::new(FsDeviceBuilder::new(dir).with_capacity(16 * MB).build().unwrap())
                 .with_block_size(MB);
@@ -1587,7 +1605,7 @@ mod tests {
 
         let e = TestError("expected unexpection".into());
 
-        let hybrid: HybridCache<u64, Vec<u8>> = HybridCacheBuilder::new()
+        let hybrid = HybridCacheBuilder::new()
             .with_name("test")
             .memory(100)
             .storage()
diff --git a/foyer/src/hybrid/writer.rs b/foyer/src/hybrid/writer.rs
index af0af867..2f84d0ef 100644
--- a/foyer/src/hybrid/writer.rs
+++ b/foyer/src/hybrid/writer.rs
@@ -21,28 +21,30 @@ use foyer_common::{
     code::{DefaultHasher, HashBuilder, StorageKey, StorageValue},
     properties::Properties,
 };
-use foyer_storage::StorageFilterResult;
+use foyer_storage::{Engine, StorageFilterResult};
 
 use crate::{HybridCache, HybridCacheEntry, HybridCacheProperties};
 
 /// Writer for hybrid cache to support more flexible write APIs.
-pub struct HybridCacheWriter<K, V, S = DefaultHasher>
+pub struct HybridCacheWriter<K, V, S = DefaultHasher, E = foyer_storage::BlockEngine<K, V, HybridCacheProperties>>
 where
     K: StorageKey,
     V: StorageValue,
     S: HashBuilder + Debug,
+    E: Engine<K, V, HybridCacheProperties>,
 {
-    hybrid: HybridCache<K, V, S>,
+    hybrid: HybridCache<K, V, S, E>,
     key: K,
 }
 
-impl<K, V, S> HybridCacheWriter<K, V, S>
+impl<K, V, S, E> HybridCacheWriter<K, V, S, E>
 where
     K: StorageKey,
     V: StorageValue,
     S: HashBuilder + Debug,
+    E: Engine<K, V, HybridCacheProperties>,
 {
-    pub(crate) fn new(hybrid: HybridCache<K, V, S>, key: K) -> Self {
+    pub(crate) fn new(hybrid: HybridCache<K, V, S, E>, key: K) -> Self {
         Self { hybrid, key }
     }
 
@@ -57,19 +59,24 @@ where
     }
 
     /// Convert [`HybridCacheWriter`] to [`HybridCacheStorageWriter`].
-    pub fn storage(self) -> HybridCacheStorageWriter<K, V, S> {
+    pub fn storage(self) -> HybridCacheStorageWriter<K, V, S, E> {
         HybridCacheStorageWriter::new(self.hybrid, self.key)
     }
 }
 
 /// Writer for disk cache of a hybrid cache to support more flexible write APIs.
-pub struct HybridCacheStorageWriter<K, V, S = DefaultHasher>
-where
+pub struct HybridCacheStorageWriter<
+    K,
+    V,
+    S = DefaultHasher,
+    E = foyer_storage::BlockEngine<K, V, HybridCacheProperties>,
+> where
     K: StorageKey,
     V: StorageValue,
     S: HashBuilder + Debug,
+    E: Engine<K, V, HybridCacheProperties>,
 {
-    hybrid: HybridCache<K, V, S>,
+    hybrid: HybridCache<K, V, S, E>,
     key: K,
     hash: u64,
 
@@ -78,13 +85,14 @@ where
     pick_duration: Duration,
 }
 
-impl<K, V, S> HybridCacheStorageWriter<K, V, S>
+impl<K, V, S, E> HybridCacheStorageWriter<K, V, S, E>
 where
     K: StorageKey,
     V: StorageValue,
     S: HashBuilder + Debug,
+    E: Engine<K, V, HybridCacheProperties>,
 {
-    pub(crate) fn new(hybrid: HybridCache<K, V, S>, key: K) -> Self {
+    pub(crate) fn new(hybrid: HybridCache<K, V, S, E>, key: K) -> Self {
         let hash = hybrid.memory().hash(&key);
         Self {
             hybrid,
diff --git a/foyer/src/prelude.rs b/foyer/src/prelude.rs
index 655e903c..be4bfb47 100644
--- a/foyer/src/prelude.rs
+++ b/foyer/src/prelude.rs
@@ -36,11 +36,11 @@ pub use crate::{
         LruConfig, S3FifoConfig, Weighter,
     },
     storage::{
-        AdmitAll, Block, BlockEngineConfig, BlockStatistics, CombinedDeviceBuilder, Compression, Device, DeviceBuilder,
-        Engine, EngineBuildContext, EngineConfig, EstimatedSize, EvictionInfo, EvictionPicker, FifoPicker,
-        FileDeviceBuilder, FsDeviceBuilder, InvalidRatioPicker, IoEngine, IoEngineConfig, IoHandle, IopsCounter, Load,
-        NoopDeviceBuilder, NoopIoEngine, NoopIoEngineConfig, PartialDeviceBuilder, PsyncIoEngine, PsyncIoEngineConfig,
-        RawFile, RecoverMode, RejectAll, Statistics, StorageFilter, StorageFilterCondition, StorageFilterResult, Store,
-        StoreBuilder, Throttle,
+        AdmitAll, Block, BlockEngine, BlockEngineConfig, BlockStatistics, CombinedDeviceBuilder, Compression, Device,
+        DeviceBuilder, Engine, EngineBuildContext, EngineConfig, EstimatedSize, EvictionInfo, EvictionPicker,
+        FifoPicker, FileDeviceBuilder, FsDeviceBuilder, InvalidRatioPicker, IoEngine, IoEngineConfig, IoHandle,
+        IopsCounter, Load, NoopDeviceBuilder, NoopEngine, NoopEngineConfig, NoopIoEngine, NoopIoEngineConfig,
+        PartialDeviceBuilder, PsyncIoEngine, PsyncIoEngineConfig, RawFile, RecoverMode, RejectAll, Statistics,
+        StorageFilter, StorageFilterCondition, StorageFilterResult, Store, StoreBuilder, Throttle,
     },
 };