Repository: Migorithm/duva
Pull request #98 (2024-12-09 20:15:35)

enhancement: DI on endecoder
Encoding and decoding logics are core domain logics with no dependency on external library. 

When required, external dependency such as tokio is injected using `TWriterFactory` trait by generic parameter on function. 






Patch:
diff --git a/src/adapters/endec/decoder/builder.rs b/src/adapters/endec/decoder/builder.rs
deleted file mode 100644
index 3abbca36..00000000
--- a/src/adapters/endec/decoder/builder.rs
+++ /dev/null
@@ -1,17 +0,0 @@
-use crate::services::statefuls::{cache::CacheEntry, persist::DatabaseSection};
-
-#[derive(Default)]
-pub struct DatabaseSectionBuilder {
-    pub(super) index: usize,
-    pub(super) storage: Vec<CacheEntry>,
-    pub(super) key_value_table_size: usize,
-    pub(super) expires_table_size: usize,
-}
-impl DatabaseSectionBuilder {
-    pub fn build(self) -> DatabaseSection {
-        DatabaseSection {
-            index: self.index,
-            storage: self.storage,
-        }
-    }
-}
diff --git a/src/adapters/io/tokio_file.rs b/src/adapters/io/tokio_file.rs
index a28b12b7..fe09f088 100644
--- a/src/adapters/io/tokio_file.rs
+++ b/src/adapters/io/tokio_file.rs
@@ -1,4 +1,7 @@
-use crate::services::query_manager::{error::IoError, interface::TWrite};
+use crate::services::query_manager::{
+    error::IoError,
+    interface::{TWrite, TWriterFactory},
+};
 use tokio::io::AsyncWriteExt;
 
 impl TWrite for tokio::fs::File {
@@ -8,3 +11,14 @@ impl TWrite for tokio::fs::File {
             .map_err(|e| e.kind().into())
     }
 }
+
+impl TWriterFactory for tokio::fs::File {
+    async fn create_writer(filepath: String) -> anyhow::Result<tokio::fs::File> {
+        tokio::fs::OpenOptions::new()
+            .write(true)
+            .create(true)
+            .open(filepath)
+            .await
+            .map_err(|e| e.into())
+    }
+}
diff --git a/src/adapters/mod.rs b/src/adapters/mod.rs
index df478930..c6505fee 100644
--- a/src/adapters/mod.rs
+++ b/src/adapters/mod.rs
@@ -1,4 +1,3 @@
 pub mod cancellation_token;
 
-pub mod endec;
 pub mod io;
diff --git a/src/lib.rs b/src/lib.rs
index d08b8062..52cc90c6 100644
--- a/src/lib.rs
+++ b/src/lib.rs
@@ -13,32 +13,31 @@ use services::query_manager::replication_request_controllers::ReplicationRequest
 use services::query_manager::QueryManager;
 use services::statefuls::cache::cache_manager::CacheManager;
 use services::statefuls::cache::ttl_manager::TtlSchedulerInbox;
-use services::statefuls::persist::endec::TEnDecoder;
+use services::statefuls::persist::persist_actor::PersistActor;
 
-pub async fn start_up<T: TCancellationTokenFactory, U: TSocketListenerFactory>(
+pub async fn start_up<C: TCancellationTokenFactory, S: TSocketListenerFactory>(
     config: Config,
 
-    endec: impl TEnDecoder,
     startup_notifier: impl TNotifyStartUp,
 ) -> Result<()> {
-    let replication_listener = U::create_listner(config.replication_bind_addr()).await;
-    let listener = U::create_listner(config.bind_addr()).await;
+    let replication_listener = S::create_listner(config.replication_bind_addr()).await;
+    let listener = S::create_listner(config.bind_addr()).await;
 
-    let (cache_manager, ttl_inbox) = CacheManager::run_cache_actors(endec);
-    cache_manager
-        .load_data(
-            ttl_inbox.clone(),
-            config.try_filepath().await,
-            config.startup_time,
-        )
-        .await?;
+    let (cache_manager, ttl_inbox) = CacheManager::run_cache_actors();
+
+    if let Some(filepath) = config.try_filepath().await {
+        let dump = PersistActor::dump(filepath).await?;
+        cache_manager
+            .dump_cache(dump, ttl_inbox.clone(), config.startup_time)
+            .await?;
+    }
 
     // Run Replication manager
     let config_manager = ConfigManager::run_actor(config);
 
     // Leak the cache_dispatcher to make it static - this is safe because the cache_dispatcher
     // will live for the entire duration of the program.
-    let cache_manager: &'static CacheManager<_> = Box::leak(Box::new(cache_manager));
+    let cache_manager: &'static CacheManager = Box::leak(Box::new(cache_manager));
 
     tokio::spawn(start_accepting_replication_connections(
         replication_listener,
@@ -46,7 +45,7 @@ pub async fn start_up<T: TCancellationTokenFactory, U: TSocketListenerFactory>(
     ));
     startup_notifier.notify_startup();
 
-    start_accepting_client_connections::<T>(listener, cache_manager, ttl_inbox, config_manager)
+    start_accepting_client_connections::<C>(listener, cache_manager, ttl_inbox, config_manager)
         .await;
     Ok(())
 }
@@ -76,13 +75,13 @@ async fn start_accepting_replication_connections(
 
 async fn start_accepting_client_connections<C: TCancellationTokenFactory>(
     listener: impl TSocketListener,
-    cache_manager: &'static CacheManager<impl TEnDecoder>,
+    cache_manager: &'static CacheManager,
     ttl_inbox: TtlSchedulerInbox,
     config_manager: ConfigManager,
 ) {
     // SAFETY: The client_request_controller is leaked to make it static.
     // This is safe because the client_request_controller will live for the entire duration of the program.
-    let client_request_controller: &'static ClientRequestController<_> =
+    let client_request_controller: &'static ClientRequestController =
         Box::leak(ClientRequestController::new(config_manager, cache_manager, ttl_inbox).into());
 
     loop {
@@ -90,10 +89,10 @@ async fn start_accepting_client_connections<C: TCancellationTokenFactory>(
             Ok((stream, _)) =>
             // Spawn a new task to handle the connection without blocking the main thread.
             {
-                tokio::spawn(QueryManager::handle_single_client_stream::<C>(
-                    stream,
-                    client_request_controller,
-                ));
+                tokio::spawn(QueryManager::handle_single_client_stream::<
+                    C,
+                    tokio::fs::File,
+                >(stream, client_request_controller));
             }
             Err(e) => {
                 if e.should_break() {
diff --git a/src/main.rs b/src/main.rs
index 7d7d7b9f..9ae58eb2 100644
--- a/src/main.rs
+++ b/src/main.rs
@@ -1,8 +1,5 @@
 use redis_starter_rust::{
-    adapters::{
-        cancellation_token::CancellationToken, endec::EnDecoder,
-        io::tokio_stream::AppStreamListener,
-    },
+    adapters::{cancellation_token::CancellationToken, io::tokio_stream::AppStreamListener},
     services::config::config_actor::Config,
     start_up,
 };
@@ -11,5 +8,5 @@ use redis_starter_rust::{
 async fn main() -> anyhow::Result<()> {
     // bootstrap dependencies
     let config = Config::default();
-    start_up::<CancellationToken, AppStreamListener>(config, EnDecoder, ()).await
+    start_up::<CancellationToken, AppStreamListener>(config, ()).await
 }
diff --git a/src/services/query_manager/client_request_controllers/mod.rs b/src/services/query_manager/client_request_controllers/mod.rs
index 75a14b79..30601fee 100644
--- a/src/services/query_manager/client_request_controllers/mod.rs
+++ b/src/services/query_manager/client_request_controllers/mod.rs
@@ -1,34 +1,29 @@
 use crate::services::config::config_manager::ConfigManager;
 use crate::services::config::ConfigResponse;
 use crate::services::query_manager::interface::TCancellationWatcher;
-
 use crate::services::query_manager::query_io::QueryIO;
 use crate::services::statefuls::cache::cache_manager::CacheManager;
 use crate::services::statefuls::cache::ttl_manager::TtlSchedulerInbox;
-use crate::services::statefuls::persist::endec::TEnDecoder;
-use crate::services::statefuls::persist::save_actor::SaveActor;
+use crate::services::statefuls::persist::endec::encoder::encoding_processor::SavingProcessor;
+use crate::services::statefuls::persist::persist_actor::PersistActor;
 use arguments::Arguments;
 use client_request::ClientRequest;
 
+use super::interface::TWriterFactory;
+
 pub mod arguments;
 pub mod client_request;
 
-pub struct ClientRequestController<U>
-where
-    U: TEnDecoder,
-{
+pub struct ClientRequestController {
     config_manager: ConfigManager,
-    cache_manager: &'static CacheManager<U>,
+    cache_manager: &'static CacheManager,
     ttl_manager: TtlSchedulerInbox,
 }
 
-impl<U> ClientRequestController<U>
-where
-    U: TEnDecoder,
-{
+impl ClientRequestController {
     pub(crate) fn new(
         config_manager: ConfigManager,
-        cache_manager: &'static CacheManager<U>,
+        cache_manager: &'static CacheManager,
         ttl_manager: TtlSchedulerInbox,
     ) -> Self {
         ClientRequestController {
@@ -38,7 +33,7 @@ where
         }
     }
 
-    pub(crate) async fn handle(
+    pub(crate) async fn handle<T: TWriterFactory>(
         &self,
         mut cancellation_token: impl TCancellationWatcher,
         cmd: ClientRequest,
@@ -61,11 +56,11 @@ where
             }
             ClientRequest::Save => {
                 // spawn save actor
-                let outbox = SaveActor::run(
+                let outbox = PersistActor::<SavingProcessor<T>>::run(
                     self.config_manager.get_filepath().await?,
                     self.cache_manager.inboxes.len(),
-                    self.cache_manager.endecoder.clone(),
-                );
+                )
+                .await?;
 
                 self.cache_manager.route_save(outbox).await;
 
diff --git a/src/services/query_manager/interface.rs b/src/services/query_manager/interface.rs
index 47b869ad..795bacc1 100644
--- a/src/services/query_manager/interface.rs
+++ b/src/services/query_manager/interface.rs
@@ -2,6 +2,8 @@ use bytes::BytesMut;
 
 use super::error::IoError;
 
+pub trait TStream: TRead + TWrite {}
+impl<T> TStream for T where T: TRead + TWrite {}
 pub trait TRead: Send + Sync + 'static {
     fn read_bytes(
         &mut self,
@@ -15,6 +17,11 @@ pub trait TWrite: Send + Sync + 'static {
         buf: &[u8],
     ) -> impl std::future::Future<Output = Result<(), IoError>> + Send;
 }
+pub trait TWriterFactory: TWrite + Send + Sync + 'static + Sized {
+    fn create_writer(
+        filepath: String,
+    ) -> impl std::future::Future<Output = anyhow::Result<Self>> + Send;
+}
 
 pub trait TCancellationTokenFactory: Send + Sync + 'static {
     fn create(timeout: u64) -> Self;
diff --git a/src/services/query_manager/mod.rs b/src/services/query_manager/mod.rs
index f9cd2e87..ddb5a935 100644
--- a/src/services/query_manager/mod.rs
+++ b/src/services/query_manager/mod.rs
@@ -1,18 +1,16 @@
 pub mod client_request_controllers;
 pub mod error;
 pub mod interface;
-
 pub mod query_io;
 pub mod replication_request_controllers;
-
 use crate::services::query_manager::client_request_controllers::client_request::ClientRequest;
 use crate::services::query_manager::client_request_controllers::ClientRequestController;
-use crate::services::statefuls::persist::endec::TEnDecoder;
+
 use anyhow::Context;
 use bytes::BytesMut;
 use client_request_controllers::arguments::Arguments;
 use error::IoError;
-use interface::{TCancellationNotifier, TCancellationTokenFactory, TRead, TWrite};
+use interface::{TCancellationNotifier, TCancellationTokenFactory, TStream, TWriterFactory};
 
 use query_io::QueryIO;
 use replication_request_controllers::ReplicationRequestController;
@@ -20,7 +18,7 @@ use replication_request_controllers::ReplicationRequestController;
 /// Controller is a struct that will be used to read and write values to the client.
 pub struct QueryManager<T, U>
 where
-    T: TWrite + TRead,
+    T: TStream,
 {
     pub(crate) stream: T,
     pub(crate) controller: U,
@@ -28,7 +26,7 @@ where
 
 impl<T, U> QueryManager<T, U>
 where
-    T: TWrite + TRead,
+    T: TStream,
 {
     pub(crate) fn new(stream: T, controller: U) -> Self {
         QueryManager { stream, controller }
@@ -49,10 +47,9 @@ where
     }
 }
 
-impl<T, U> QueryManager<T, &'static ClientRequestController<U>>
+impl<T> QueryManager<T, &'static ClientRequestController>
 where
-    T: TWrite + TRead,
-    U: TEnDecoder,
+    T: TStream,
 {
     async fn extract_query(&mut self) -> anyhow::Result<(ClientRequest, Arguments)> {
         let query_io = self.read_value().await?;
@@ -71,9 +68,9 @@ where
         }
     }
 
-    pub async fn handle_single_client_stream<C: TCancellationTokenFactory>(
+    pub async fn handle_single_client_stream<C: TCancellationTokenFactory, F: TWriterFactory>(
         stream: T,
-        controller: &'static ClientRequestController<U>,
+        controller: &'static ClientRequestController,
     ) {
         const TIMEOUT: u64 = 100;
         let mut query_manager = QueryManager::new(stream, controller);
@@ -92,7 +89,7 @@ where
 
             let res = match query_manager
                 .controller
-                .handle(cancellation_token, request, query_args)
+                .handle::<F>(cancellation_token, request, query_args)
                 .await
             {
                 Ok(response) => query_manager.write_value(response).await,
@@ -110,7 +107,7 @@ where
 
 impl<T> QueryManager<T, ReplicationRequestController>
 where
-    T: TWrite + TRead,
+    T: TStream,
 {
     pub(crate) async fn handle_replica_stream(self) -> Result<(), std::io::Error> {
         Ok(())
diff --git a/src/services/statefuls/cache/cache_actor.rs b/src/services/statefuls/cache/cache_actor.rs
index 7b459111..ed2124af 100644
--- a/src/services/statefuls/cache/cache_actor.rs
+++ b/src/services/statefuls/cache/cache_actor.rs
@@ -3,7 +3,7 @@ use super::CacheEntry;
 use super::CacheValue;
 use crate::make_smart_pointer;
 use crate::services::query_manager::query_io::QueryIO;
-use crate::services::statefuls::persist::save_actor::SaveActorCommand;
+use crate::services::statefuls::persist::persist_actor::SaveCommand;
 use anyhow::Context;
 use anyhow::Result;
 use std::collections::HashMap;
@@ -15,7 +15,7 @@ pub enum CacheCommand {
         ttl_sender: TtlSchedulerInbox,
     },
     Save {
-        outbox: mpsc::Sender<SaveActorCommand>,
+        outbox: mpsc::Sender<SaveCommand>,
     },
     Get {
         key: String,
@@ -74,7 +74,6 @@ impl CacheActor {
                 CacheCommand::Keys { pattern, sender } => {
                     let ks: Vec<_> = self.keys_stream(pattern).collect();
 
-                    eprintln!("ks.len() = {:?}", ks.len());
                     sender
                         .send(QueryIO::Array(ks))
                         .map_err(|_| anyhow::anyhow!("Error sending keys"))?;
@@ -84,18 +83,18 @@ impl CacheActor {
                 }
                 CacheCommand::Save { outbox } => {
                     outbox
-                        .send(SaveActorCommand::LocalShardSize {
+                        .send(SaveCommand::LocalShardSize {
                             table_size: self.len(),
                             expiry_size: self.keys_with_expiry(),
                         })
                         .await?;
                     for chunk in self.cache.iter().collect::<Vec<_>>().chunks(10) {
                         outbox
-                            .send(SaveActorCommand::SaveChunk(CacheEntry::new(chunk)))
+                            .send(SaveCommand::SaveChunk(CacheEntry::new(chunk)))
                             .await?;
                     }
                     // finalize the save operation
-                    outbox.send(SaveActorCommand::StopSentinel).await?;
+                    outbox.send(SaveCommand::StopSentinel).await?;
                 }
             }
         }
diff --git a/src/services/statefuls/cache/cache_manager.rs b/src/services/statefuls/cache/cache_manager.rs
index 20e39ecc..2688dec5 100644
--- a/src/services/statefuls/cache/cache_manager.rs
+++ b/src/services/statefuls/cache/cache_manager.rs
@@ -2,8 +2,9 @@ use super::cache_actor::{CacheActor, CacheCommand, CacheCommandSender};
 use super::ttl_manager::{TtlActor, TtlSchedulerInbox};
 use super::CacheEntry;
 use crate::services::query_manager::query_io::QueryIO;
-use crate::services::statefuls::persist::endec::TEnDecoder;
-use crate::services::statefuls::persist::save_actor::SaveActorCommand;
+
+use crate::services::statefuls::persist::persist_actor::SaveCommand;
+use crate::services::statefuls::persist::DumpFile;
 use anyhow::Result;
 use std::time::SystemTime;
 use std::{hash::Hasher, iter::Zip};
@@ -15,26 +16,19 @@ type OneShotReceiverJoinHandle<T> =
     tokio::task::JoinHandle<std::result::Result<T, tokio::sync::oneshot::error::RecvError>>;
 
 #[derive(Clone)]
-pub(crate) struct CacheManager<EnDec> {
+pub(crate) struct CacheManager {
     pub(crate) inboxes: Vec<CacheCommandSender>,
-    pub(crate) endecoder: EnDec,
 }
 
-impl<EnDec: TEnDecoder> CacheManager<EnDec> {
-    pub(crate) async fn load_data(
+impl CacheManager {
+    pub(crate) async fn dump_cache(
         &self,
+        dump: DumpFile,
         ttl_inbox: TtlSchedulerInbox,
-        filepath: Option<String>,
         startup_time: SystemTime,
     ) -> Result<()> {
-        let Some(filepath) = filepath else {
-            return Ok(());
-        };
-        let bytes = tokio::fs::read(filepath).await?;
-        let database = self.endecoder.decode_data(bytes)?;
-
         let startup_time = &startup_time;
-        for kvs in database
+        for kvs in dump
             .key_values()
             .into_iter()
             .filter(|kvs| kvs.is_valid(startup_time))
@@ -68,7 +62,7 @@ impl<EnDec: TEnDecoder> CacheManager<EnDec> {
         Ok(QueryIO::SimpleString("OK".to_string()))
     }
 
-    pub(crate) async fn route_save(&self, outbox: mpsc::Sender<SaveActorCommand>) {
+    pub(crate) async fn route_save(&self, outbox: mpsc::Sender<SaveCommand>) {
         // get all the handlers to cache actors
         for inbox in self.inboxes.iter().map(Clone::clone) {
             let outbox = outbox.clone();
@@ -102,16 +96,12 @@ impl<EnDec: TEnDecoder> CacheManager<EnDec> {
                 sender,
             ));
         });
-
         let mut keys = Vec::new();
         for v in receivers {
             if let Ok(QueryIO::Array(v)) = v.await? {
                 keys.extend(v)
             }
         }
-
-        eprintln!("keys.len() = {:?}", keys.len());
-
         Ok(QueryIO::Array(keys))
     }
 
@@ -147,14 +137,13 @@ impl<EnDec: TEnDecoder> CacheManager<EnDec> {
         hasher.finish() as usize % self.inboxes.len()
     }
 
-    pub fn run_cache_actors(decoder: EnDec) -> (CacheManager<EnDec>, TtlSchedulerInbox) {
+    pub fn run_cache_actors() -> (CacheManager, TtlSchedulerInbox) {
         const NUM_OF_PERSISTENCE: usize = 10;
 
         let cache_dispatcher = CacheManager {
             inboxes: (0..NUM_OF_PERSISTENCE)
                 .map(|_| CacheActor::run())
                 .collect::<Vec<_>>(),
-            endecoder: decoder,
         };
 
         let ttl_inbox = cache_dispatcher.run_ttl_actors();
diff --git a/src/services/statefuls/cache/ttl_manager.rs b/src/services/statefuls/cache/ttl_manager.rs
index fa39c1ea..380a818a 100644
--- a/src/services/statefuls/cache/ttl_manager.rs
+++ b/src/services/statefuls/cache/ttl_manager.rs
@@ -1,9 +1,6 @@
 use crate::{
     make_smart_pointer,
-    services::statefuls::{
-        cache::{cache_actor::CacheCommand, cache_manager::CacheManager},
-        persist::endec::TEnDecoder,
-    },
+    services::statefuls::cache::{cache_actor::CacheCommand, cache_manager::CacheManager},
 };
 use std::{
     cmp::Reverse,
@@ -21,7 +18,7 @@ use tokio::{
 pub struct TtlActor;
 
 impl TtlActor {
-    pub(crate) fn run<T: TEnDecoder>(cache_dispatcher: CacheManager<T>) -> TtlSchedulerInbox {
+    pub(crate) fn run(cache_dispatcher: CacheManager) -> TtlSchedulerInbox {
         let (scheduler_outbox, inbox) = tokio::sync::mpsc::channel(100);
         tokio::spawn(Self::ttl_schedule_actor(inbox));
         tokio::spawn(Self::background_delete_actor(
@@ -33,8 +30,8 @@ impl TtlActor {
     }
 
     // Background actor keeps sending peek command to the scheduler actor to check if there is any key to delete.
-    async fn background_delete_actor<T: TEnDecoder>(
-        cache_manager: CacheManager<T>,
+    async fn background_delete_actor(
+        cache_manager: CacheManager,
         outbox: mpsc::Sender<TtlCommand>,
     ) -> anyhow::Result<()> {
         let mut cleanup_interval = interval(Duration::from_millis(1));
diff --git a/src/services/statefuls/persist/endec.rs b/src/services/statefuls/persist/endec.rs
deleted file mode 100644
index b7c9e1fd..00000000
--- a/src/services/statefuls/persist/endec.rs
+++ /dev/null
@@ -1,26 +0,0 @@
-use crate::services::{
-    interfaces::ThreadSafeCloneable,
-    statefuls::persist::{save_actor::SaveActorCommand, RdbFile},
-};
-
-pub trait TEnDecoder: TDecodeData + TEncodeData {}
-impl<T: TDecodeData + TEncodeData> TEnDecoder for T {}
-pub trait TDecodeData: ThreadSafeCloneable {
-    fn decode_data(&self, bytes: Vec<u8>) -> anyhow::Result<RdbFile>;
-}
-
-pub trait TEncodeData: ThreadSafeCloneable {
-    fn create_encoding_processor(
-        &self,
-        filepath: &str,
-        number_of_cache_actors: usize,
-    ) -> impl std::future::Future<Output = anyhow::Result<impl TEncodingProcessor>> + Send;
-}
-
-pub trait TEncodingProcessor: Send + Sync {
-    fn add_meta(&mut self) -> impl std::future::Future<Output = anyhow::Result<()>> + Send;
-    fn handle_cmd(
-        &mut self,
-        cmd: SaveActorCommand,
-    ) -> impl std::future::Future<Output = anyhow::Result<bool>> + Send;
-}
diff --git a/src/adapters/endec/decoder/byte_decoder.rs b/src/services/statefuls/persist/endec/decoder/byte_decoder.rs
similarity index 99%
rename from src/adapters/endec/decoder/byte_decoder.rs
rename to src/services/statefuls/persist/endec/decoder/byte_decoder.rs
index 052f43e3..26d006c4 100644
--- a/src/adapters/endec/decoder/byte_decoder.rs
+++ b/src/services/statefuls/persist/endec/decoder/byte_decoder.rs
@@ -1,10 +1,10 @@
-use crate::adapters::endec::{
+use crate::services::statefuls::cache::CacheEntry;
+use crate::services::statefuls::persist::endec::{
     extract_range, StoredDuration, DATABASE_SECTION_INDICATOR, DATABASE_TABLE_SIZE_INDICATOR,
     EXPIRY_TIME_IN_MILLISECONDS_INDICATOR, EXPIRY_TIME_IN_SECONDS_INDICATOR, HEADER_MAGIC_STRING,
     METADATA_SECTION_INDICATOR, STRING_VALUE_TYPE_INDICATOR,
 };
-use crate::services::statefuls::cache::CacheEntry;
-use crate::services::statefuls::persist::{DatabaseSection, RdbFile};
+use crate::services::statefuls::persist::{DatabaseSection, DatabaseSectionBuilder, DumpFile};
 
 use anyhow::{Context, Result};
 
@@ -14,7 +14,6 @@ use std::{
     time::SystemTime,
 };
 
-use super::builder::DatabaseSectionBuilder;
 use super::states::{DecoderInit, HeaderReady, MetadataReady};
 
 #[derive(Default)]
@@ -181,7 +180,7 @@ impl<'a> BytesDecoder<'a, HeaderReady> {
     }
 }
 impl BytesDecoder<'_, MetadataReady> {
-    pub fn load_database(mut self) -> Result<RdbFile> {
+    pub fn load_database(mut self) -> Result<DumpFile> {
         let mut database = Vec::new();
         while self.check_indicator(DATABASE_SECTION_INDICATOR) {
             let section = self
@@ -192,7 +191,7 @@ impl BytesDecoder<'_, MetadataReady> {
         }
 
         let checksum = self.try_get_checksum()?;
-        Ok(RdbFile {
+        Ok(DumpFile {
             header: self.state.header,
             metadata: self.state.metadata,
             database,
diff --git a/src/adapters/endec/decoder/mod.rs b/src/services/statefuls/persist/endec/decoder/mod.rs
similarity index 74%
rename from src/adapters/endec/decoder/mod.rs
rename to src/services/statefuls/persist/endec/decoder/mod.rs
index c6a03022..3b11b798 100644
--- a/src/adapters/endec/decoder/mod.rs
+++ b/src/services/statefuls/persist/endec/decoder/mod.rs
@@ -1,3 +1,3 @@
-mod builder;
 pub mod byte_decoder;
+
 pub mod states;
diff --git a/src/adapters/endec/decoder/states.rs b/src/services/statefuls/persist/endec/decoder/states.rs
similarity index 100%
rename from src/adapters/endec/decoder/states.rs
rename to src/services/statefuls/persist/endec/decoder/states.rs
diff --git a/src/adapters/endec/encoder/byte_encoder.rs b/src/services/statefuls/persist/endec/encoder/byte_encoder.rs
similarity index 97%
rename from src/adapters/endec/encoder/byte_encoder.rs
rename to src/services/statefuls/persist/endec/encoder/byte_encoder.rs
index ed8b60b4..7b961ced 100644
--- a/src/adapters/endec/encoder/byte_encoder.rs
+++ b/src/services/statefuls/persist/endec/encoder/byte_encoder.rs
@@ -1,10 +1,10 @@
-use crate::{
-    adapters::endec::{
+use crate::services::statefuls::{
+    cache::CacheEntry,
+    persist::endec::{
         CHECKSUM_INDICATOR, DATABASE_SECTION_INDICATOR, DATABASE_TABLE_SIZE_INDICATOR,
         EXPIRY_TIME_IN_MILLISECONDS_INDICATOR, HEADER_MAGIC_STRING, METADATA_SECTION_INDICATOR,
         STRING_VALUE_TYPE_INDICATOR,
     },
-    services::statefuls::cache::CacheEntry,
 };
 
 use anyhow::Result;
@@ -132,9 +132,12 @@ fn encode_size(size: usize) -> Result<Vec<u8>> {
 
 #[cfg(test)]
 mod test {
+    use crate::services::statefuls::persist::endec::{
+        decoder::{byte_decoder::BytesDecoder, states::DecoderInit},
+        StoredDuration,
+    };
+
     use super::*;
-    use crate::adapters::endec::decoder::byte_decoder::BytesDecoder;
-    use crate::adapters::endec::decoder::states::DecoderInit;
 
     #[test]
     fn test_size_encode_6_bit() {
@@ -327,7 +330,6 @@ mod test {
 
     #[test]
     fn test_cache_value_with_expiry_milliseconds() {
-        use crate::adapters::endec::StoredDuration;
         let kvs = CacheEntry::KeyValueExpiry(
             "key".to_string(),
             "value".to_string(),
diff --git a/src/adapters/endec/encoder/encoding_processor.rs b/src/services/statefuls/persist/endec/encoder/encoding_processor.rs
similarity index 76%
rename from src/adapters/endec/encoder/encoding_processor.rs
rename to src/services/statefuls/persist/endec/encoder/encoding_processor.rs
index 499ec972..3da37989 100644
--- a/src/adapters/endec/encoder/encoding_processor.rs
+++ b/src/services/statefuls/persist/endec/encoder/encoding_processor.rs
@@ -1,30 +1,27 @@
-use crate::adapters::endec::byte_encoder::{
-    encode_checksum, encode_database_info, encode_database_table_size, encode_header,
-    encode_metadata,
-};
 use crate::services::query_manager::interface::TWrite;
 use crate::services::statefuls::cache::CacheEntry;
-use crate::services::statefuls::persist::endec::TEncodingProcessor;
-use crate::services::statefuls::persist::save_actor::SaveActorCommand;
+
+use crate::services::statefuls::persist::persist_actor::SaveCommand;
 use anyhow::Result;
 
 use std::collections::VecDeque;
 
-pub struct EncodingProcessor<T: TWrite> {
+use super::byte_encoder::{
+    encode_checksum, encode_database_info, encode_database_table_size, encode_header,
+    encode_metadata,
+};
+
+pub struct SavingProcessor<T: TWrite> {
     pub(super) writer: T,
-    pub(super) meta: EncodingMeta,
+    pub(super) meta: SaveMeta,
 }
 
-pub(crate) struct EncodingMeta {
-    num_of_saved_table_size_actor: usize,
-    total_key_value_table_size: usize,
-    total_expires_table_size: usize,
-    chunk_queue: VecDeque<Vec<CacheEntry>>,
-    num_of_cache_actors: usize,
-}
+impl<T: TWrite> SavingProcessor<T> {
+    pub fn new(file: T, meta: SaveMeta) -> Self {
+        Self { writer: file, meta }
+    }
 
-impl<T: TWrite> TEncodingProcessor for EncodingProcessor<T> {
-    async fn add_meta(&mut self) -> Result<()> {
+    pub async fn add_meta(&mut self) -> Result<()> {
         let meta = [
             encode_header("0011")?,
             encode_metadata(Vec::from([("redis-ver", "6.0.16")]))?,
@@ -33,9 +30,9 @@ impl<T: TWrite> TEncodingProcessor for EncodingProcessor<T> {
         self.writer.write_all(&meta.concat()).await?;
         Ok(())
     }
-    async fn handle_cmd(&mut self, cmd: SaveActorCommand) -> Result<bool> {
+    pub async fn handle_cmd(&mut self, cmd: SaveCommand) -> Result<bool> {
         match cmd {
-            SaveActorCommand::LocalShardSize {
+            SaveCommand::LocalShardSize {
                 table_size,
                 expiry_size,
             } => {
@@ -51,13 +48,13 @@ impl<T: TWrite> TEncodingProcessor for EncodingProcessor<T> {
                         .await?;
                 }
             }
-            SaveActorCommand::SaveChunk(chunk) => {
+            SaveCommand::SaveChunk(chunk) => {
                 self.meta.chunk_queue.push_back(chunk);
                 if self.meta.num_of_saved_table_size_actor == 0 {
                     self.encode_chunk_queue().await?;
                 }
             }
-            SaveActorCommand::StopSentinel => {
+            SaveCommand::StopSentinel => {
                 self.meta.num_of_cache_actors -= 1;
                 if self.meta.num_of_cache_actors == 0 {
                     self.encode_chunk_queue().await?;
@@ -69,12 +66,6 @@ impl<T: TWrite> TEncodingProcessor for EncodingProcessor<T> {
         }
         Ok(false)
     }
-}
-
-impl<T: TWrite> EncodingProcessor<T> {
-    pub fn new(file: T, meta: EncodingMeta) -> Self {
-        Self { writer: file, meta }
-    }
 
     async fn encode_chunk_queue(&mut self) -> Result<()> {
         while let Some(chunk) = self.meta.chunk_queue.pop_front() {
@@ -87,8 +78,16 @@ impl<T: TWrite> EncodingProcessor<T> {
     }
 }
 
-impl EncodingMeta {
-    pub(in crate::adapters::endec) fn new(num_of_cache_actors: usize) -> Self {
+pub struct SaveMeta {
+    num_of_saved_table_size_actor: usize,
+    total_key_value_table_size: usize,
+    total_expires_table_size: usize,
+    chunk_queue: VecDeque<Vec<CacheEntry>>,
+    num_of_cache_actors: usize,
+}
+
+impl SaveMeta {
+    pub fn new(num_of_cache_actors: usize) -> Self {
         Self {
             num_of_saved_table_size_actor: num_of_cache_actors,
             total_key_value_table_size: 0,
diff --git a/src/adapters/endec/encoder/mod.rs b/src/services/statefuls/persist/endec/encoder/mod.rs
similarity index 100%
rename from src/adapters/endec/encoder/mod.rs
rename to src/services/statefuls/persist/endec/encoder/mod.rs
diff --git a/src/adapters/endec/mod.rs b/src/services/statefuls/persist/endec/mod.rs
similarity index 77%
rename from src/adapters/endec/mod.rs
rename to src/services/statefuls/persist/endec/mod.rs
index 54e6a78e..f48fb617 100644
--- a/src/adapters/endec/mod.rs
+++ b/src/services/statefuls/persist/endec/mod.rs
@@ -79,13 +79,6 @@ use std::time::SystemTime;
 pub mod decoder;
 pub mod encoder;
 
-use decoder::byte_decoder::BytesDecoder;
-use decoder::states::DecoderInit;
-pub use encoder::byte_encoder;
-use encoder::encoding_processor::{EncodingMeta, EncodingProcessor};
-
-use crate::services::statefuls::persist::endec::{TDecodeData, TEncodeData, TEncodingProcessor};
-
 const HEADER_MAGIC_STRING: &str = "REDIS";
 const METADATA_SECTION_INDICATOR: u8 = 0xFA;
 const DATABASE_SECTION_INDICATOR: u8 = 0xFE;
@@ -117,36 +110,3 @@ impl StoredDuration {
         }
     }
 }
-
-#[derive(Default, Debug, Clone)]
-pub struct EnDecoder;
-
-impl TDecodeData for EnDecoder {
-    fn decode_data(
-        &self,
-        bytes: Vec<u8>,
-    ) -> anyhow::Result<crate::services::statefuls::persist::RdbFile> {
-        let decoder: BytesDecoder<DecoderInit> = bytes.as_slice().into();
-        let database = decoder.load_header()?.load_metadata()?.load_database()?;
-        Ok(database)
-    }
-}
-
-impl TEncodeData for EnDecoder {
-    async fn create_encoding_processor(
-        &self,
-        filepath: &str,
-        num_of_cache_actors: usize,
-    ) -> anyhow::Result<impl TEncodingProcessor> {
-        let file = tokio::fs::OpenOptions::new()
-            .write(true)
-            .create(true)
-            .open(filepath)
-            .await?;
-
-        Ok(EncodingProcessor::new(
-            file,
-            EncodingMeta::new(num_of_cache_actors),
-        ))
-    }
-}
diff --git a/src/services/statefuls/persist/mod.rs b/src/services/statefuls/persist/mod.rs
index ce276f7a..79f313a5 100644
--- a/src/services/statefuls/persist/mod.rs
+++ b/src/services/statefuls/persist/mod.rs
@@ -1,19 +1,19 @@
 pub mod endec;
-pub mod save_actor;
+pub mod persist_actor;
 
 use std::collections::HashMap;
 
 use super::cache::CacheEntry;
 
 #[derive(Debug)]
-pub struct RdbFile {
+pub struct DumpFile {
     pub(crate) header: String,
     pub(crate) metadata: HashMap<String, String>,
     pub(crate) database: Vec<DatabaseSection>,
     pub(crate) checksum: Vec<u8>,
 }
 
-impl RdbFile {
+impl DumpFile {
     pub fn new(
         header: String,
         metadata: HashMap<String, String>,
@@ -40,3 +40,19 @@ pub struct DatabaseSection {
     pub index: usize,
     pub storage: Vec<CacheEntry>,
 }
+
+#[derive(Default)]
+pub(crate) struct DatabaseSectionBuilder {
+    pub(crate) index: usize,
+    pub(crate) storage: Vec<CacheEntry>,
+    pub(crate) key_value_table_size: usize,
+    pub(crate) expires_table_size: usize,
+}
+impl DatabaseSectionBuilder {
+    pub fn build(self) -> DatabaseSection {
+        DatabaseSection {
+            index: self.index,
+            storage: self.storage,
+        }
+    }
+}
diff --git a/src/services/statefuls/persist/persist_actor.rs b/src/services/statefuls/persist/persist_actor.rs
new file mode 100644
index 00000000..ae22fd7c
--- /dev/null
+++ b/src/services/statefuls/persist/persist_actor.rs
@@ -0,0 +1,73 @@
+use super::endec::decoder::byte_decoder::BytesDecoder;
+use super::endec::decoder::states::DecoderInit;
+use super::endec::encoder::encoding_processor::SaveMeta;
+use super::endec::encoder::encoding_processor::SavingProcessor;
+use super::DumpFile;
+use crate::services::query_manager::interface::TWriterFactory;
+use crate::services::statefuls::cache::CacheEntry;
+use tokio::sync::mpsc::{Receiver, Sender};
+
+pub enum SaveCommand {
+    LocalShardSize {
+        table_size: usize,
+        expiry_size: usize,
+    },
+    SaveChunk(Vec<CacheEntry>),
+    StopSentinel,
+}
+
+pub struct Load;
+
+pub struct PersistActor<T> {
+    processor: T,
+}
+
+impl PersistActor<Load> {
+    pub(crate) async fn dump(filepath: String) -> anyhow::Result<DumpFile> {
+        let bytes = tokio::fs::read(filepath).await?;
+
+        let decoder: BytesDecoder<DecoderInit> = bytes.as_slice().into();
+        let database = decoder.load_header()?.load_metadata()?.load_database()?;
+        Ok(database)
+    }
+}
+
+impl<T> PersistActor<SavingProcessor<T>>
+where
+    T: TWriterFactory,
+{
+    pub async fn run(
+        filepath: String,
+        num_of_cache_actors: usize,
+    ) -> anyhow::Result<Sender<SaveCommand>> {
+        // * Propagate error to caller before sending it to the background
+        let file = T::create_writer(filepath).await?;
+
+        let processor = SavingProcessor::new(file, SaveMeta::new(num_of_cache_actors));
+
+        let actor = Self { processor };
+
+        let (outbox, inbox) = tokio::sync::mpsc::channel(100);
+
+        tokio::spawn(actor.save(inbox));
+        Ok(outbox)
+    }
+
+    async fn save(mut self, mut inbox: Receiver<SaveCommand>) -> anyhow::Result<()> {
+        self.processor.add_meta().await?;
+        while let Some(command) = inbox.recv().await {
+            match self.processor.handle_cmd(command).await {
+                Ok(should_break) => {
+                    if should_break {
+                        break;
+                    }
+                }
+                Err(err) => {
+                    eprintln!("error while encoding: {:?}", err);
+                    return Err(err);
+                }
+            }
+        }
+        Ok(())
+    }
+}
diff --git a/src/services/statefuls/persist/save_actor.rs b/src/services/statefuls/persist/save_actor.rs
deleted file mode 100644
index a52b9d8b..00000000
--- a/src/services/statefuls/persist/save_actor.rs
+++ /dev/null
@@ -1,56 +0,0 @@
-use crate::services::statefuls::{cache::CacheEntry, persist::endec::TEncodingProcessor};
-use tokio::sync::mpsc::{Receiver, Sender};
-
-use super::endec::TEncodeData;
-
-pub enum SaveActorCommand {
-    LocalShardSize {
-        table_size: usize,
-        expiry_size: usize,
-    },
-    SaveChunk(Vec<CacheEntry>),
-    StopSentinel,
-}
-
-pub struct SaveActor;
-
-impl SaveActor {
-    pub fn run(
-        filepath: String,
-        num_of_cache_actors: usize,
-        // TODO encoder seems to work as actual save actor.
-        encoder: impl TEncodeData,
-    ) -> Sender<SaveActorCommand> {
-        let (outbox, inbox) = tokio::sync::mpsc::channel(100);
-
-        tokio::spawn(Self::handle(encoder, filepath, num_of_cache_actors, inbox));
-        outbox
-    }
-
-    async fn handle(
-        encoder: impl TEncodeData,
-        filepath: String,
-        number_of_cache_actors: usize,
-        mut inbox: Receiver<SaveActorCommand>,
-    ) -> anyhow::Result<()> {
-        let mut processor = encoder
-            .create_encoding_processor(&filepath, number_of_cache_actors)
-            .await?;
-
-        processor.add_meta().await?;
-        while let Some(command) = inbox.recv().await {
-            match processor.handle_cmd(command).await {
-                Ok(should_break) => {
-                    if should_break {
-                        break;
-                    }
-                }
-                Err(err) => {
-                    eprintln!("error while encoding: {:?}", err);
-                    return Err(err);
-                }
-            }
-        }
-        Ok(())
-    }
-}


Test patch:
diff --git a/tests/common.rs b/tests/common.rs
index 591e7cc4..dce54c28 100644
--- a/tests/common.rs
+++ b/tests/common.rs
@@ -77,9 +77,7 @@ pub async fn start_test_server<T: TCancellationTokenFactory>(
     let start_flag = StartFlag(notify.clone());
 
     let h = tokio::spawn(redis_starter_rust::start_up::<T, AppStreamListener>(
-        config,
-        redis_starter_rust::adapters::endec::EnDecoder,
-        start_flag,
+        config, start_flag,
     ));
 
     //warm up time
diff --git a/tests/test_save_read_dump.rs b/tests/test_save_read_dump.rs
index 9c122a53..66338022 100644
--- a/tests/test_save_read_dump.rs
+++ b/tests/test_save_read_dump.rs
@@ -13,14 +13,22 @@ use redis_starter_rust::adapters::cancellation_token::CancellationToken;
 use std::time::{SystemTime, UNIX_EPOCH};
 use tokio::net::TcpStream;
 
+struct FileName(String);
+impl Drop for FileName {
+    fn drop(&mut self) {
+        std::fs::remove_file(&self.0).unwrap();
+    }
+}
+
+// TODO response cannot be deterministic!
 #[tokio::test]
 async fn test_save_read_dump() {
     // GIVEN
-    let test_file_name = create_unique_file_name("test_save_dump");
+    let test_file_name = FileName(create_unique_file_name("test_save_dump"));
     let mut config = init_config_with_free_port().await;
     let old_ref = config.dbfilename;
 
-    config.dbfilename = Box::leak(Box::new(test_file_name.clone()));
+    config.dbfilename = Box::leak(Box::new(test_file_name.0.clone()));
     unsafe {
         let ptr: *const str = old_ref;
         let _ = Box::from_raw(ptr as *mut str);
@@ -41,7 +49,7 @@ async fn test_save_read_dump() {
     assert_eq!(h.get_response().await, ok_response());
     // check keys
     h.send(keys_command("*").as_slice()).await;
-    assert_eq!(h.get_response().await, array(vec!["foo", "foo2"]));
+    assert_eq!(h.get_response().await, array(vec!["foo2", "foo"]));
     // save
     h.send(save_command().as_slice()).await;
 
@@ -59,10 +67,7 @@ async fn test_save_read_dump() {
 
     // keys
     h.send(keys_command("*").as_slice()).await;
-    assert_eq!(h.get_response().await, array(vec!["foo", "foo2"]));
-
-    // remove file
-    std::fs::remove_file(&test_file_name).unwrap();
+    assert_eq!(h.get_response().await, array(vec!["foo2", "foo"]));
 }
 
 fn create_unique_file_name(function_name: &str) -> String {