Repository: Migorithm/duva
Pull request #158 (2025-01-20 18:47:35)

feat/sync-full-file-to-replica
### Pull Request Description

**Title:** feat/sync-full-file-to-replica

**Overview:**
This pull request introduces several enhancements and refactorings aimed at improving the synchronization of full files to the replica. The changes include the addition of new features, method extractions, and improvements to the code structure for better readability and maintainability.

**Changes Made:**
- **Outline for Sync:** Established a foundational outline for the synchronization process.
- **Encoding Target:** Implemented a new encoding target to facilitate file encoding during the sync process.
- **Refactor:** Renamed `save_command` to `encoding_command` for clarity and consistency in naming conventions.
- **New File Type:** Introduced `QueryIo::File` type to better represent file operations within the sync context.
- **Hex Parsing:** Implemented parsing and reading of hex data to support various file formats and data types.
- **Testing:** Added tests to validate the sync functionality, ensuring reliability and correctness of the new features.



Patch:
diff --git a/src/lib.rs b/src/lib.rs
index afda2371..a8747058 100644
--- a/src/lib.rs
+++ b/src/lib.rs
@@ -83,6 +83,7 @@ where
         tokio::spawn(Self::start_accepting_peer_connections(
             self.config_manager.peer_bind_addr(),
             self.cluster_manager,
+            self.cache_manager,
         ));
 
         tokio::spawn(async move {
@@ -96,6 +97,7 @@ where
     async fn start_accepting_peer_connections(
         peer_bind_addr: String,
         cluster_manager: &'static ClusterManager,
+        cache_manager: &'static CacheManager,
     ) -> Result<()> {
         let peer_listener = TcpListener::bind(&peer_bind_addr)
             .await
@@ -108,9 +110,15 @@ where
             match peer_listener.accept().await {
                 // ? how do we know if incoming connection is from a peer or replica?
                 Ok((peer_stream, _socket_addr)) => {
+                    let repl_info = cluster_manager.replication_info().await?;
+
                     tokio::spawn(async move {
-                        if let Err(err) =
-                            cluster_manager.accept_peer(InboundStream(peer_stream)).await
+                        if let Err(err) = cluster_manager
+                            .accept_inbound_stream(
+                                InboundStream::new(peer_stream, repl_info),
+                                cache_manager,
+                            )
+                            .await
                         {
                             println!("[ERROR] Failed to accept peer connection: {:?}", err);
                         }
diff --git a/src/macros/mod.rs b/src/macros/mod.rs
index 1310ada5..faddb573 100644
--- a/src/macros/mod.rs
+++ b/src/macros/mod.rs
@@ -50,5 +50,10 @@ macro_rules! from_to {
                 Self(value)
             }
         }
+        impl From<$to> for $from {
+            fn from(value: $to) -> Self {
+                value.0
+            }
+        }
     };
 }
diff --git a/src/services/client/manager.rs b/src/services/client/manager.rs
index c1bce3b0..06a14eb7 100644
--- a/src/services/client/manager.rs
+++ b/src/services/client/manager.rs
@@ -12,7 +12,7 @@ use crate::services::query_io::QueryIO;
 use crate::services::statefuls::cache::manager::CacheManager;
 use crate::services::statefuls::cache::ttl::manager::TtlSchedulerInbox;
 use crate::services::statefuls::persist::actor::PersistActor;
-use crate::services::statefuls::persist::endec::encoder::encoding_processor::SavingProcessor;
+use crate::services::statefuls::persist::endec::encoder::encoding_processor::EncodingProcessor;
 use tokio::net::TcpListener;
 use tokio::select;
 
@@ -54,7 +54,7 @@ impl ClientManager {
             }
             ClientRequest::Save => {
                 // spawn save actor
-                let outbox = PersistActor::<SavingProcessor>::run(
+                let outbox = PersistActor::<EncodingProcessor>::run(
                     self.config_manager.get_filepath().await?,
                     self.cache_manager.inboxes.len(),
                 )
diff --git a/src/services/cluster/actors/actor.rs b/src/services/cluster/actors/actor.rs
index 013a0c72..21aaec49 100644
--- a/src/services/cluster/actors/actor.rs
+++ b/src/services/cluster/actors/actor.rs
@@ -69,7 +69,7 @@ impl ClusterActor {
 
                 ClusterCommand::GetPeers(callback) => {
                     // send
-                    let _ = callback.send(self.members.keys().cloned().collect());
+                    let _ = callback.send(self.members.keys().cloned().collect::<Vec<_>>().into());
                 }
                 ClusterCommand::Write(write_cmd) => match write_cmd {
                     ClusterWriteCommand::Replicate { query: _ } => todo!(),
diff --git a/src/services/cluster/actors/command.rs b/src/services/cluster/actors/command.rs
index 99e9a199..19f3b128 100644
--- a/src/services/cluster/actors/command.rs
+++ b/src/services/cluster/actors/command.rs
@@ -3,13 +3,13 @@ use tokio::net::TcpStream;
 
 use super::{
     replication::Replication,
-    types::{PeerAddr, PeerKind},
+    types::{PeerAddr, PeerAddrs, PeerKind},
 };
 
 pub enum ClusterCommand {
     AddPeer { peer_addr: PeerAddr, stream: TcpStream, peer_kind: PeerKind },
     RemovePeer(PeerAddr),
-    GetPeers(tokio::sync::oneshot::Sender<Vec<PeerAddr>>),
+    GetPeers(tokio::sync::oneshot::Sender<PeerAddrs>),
     ReplicationInfo(tokio::sync::oneshot::Sender<Replication>),
     SetReplicationId(String),
     Write(ClusterWriteCommand),
@@ -26,9 +26,11 @@ pub enum ClusterWriteCommand {
     Ping,
 }
 
+#[derive(Debug)]
 pub enum MasterCommand {
     Ping,
     Replicate { query: QueryIO },
+    Sync(QueryIO),
 }
 pub enum SlaveCommand {
     Ping,
@@ -42,9 +44,9 @@ impl TryFrom<QueryIO> for MasterCommand {
     type Error = anyhow::Error;
     fn try_from(query: QueryIO) -> anyhow::Result<Self> {
         match query {
+            QueryIO::File(v) => Ok(Self::Sync(v.into())),
             QueryIO::SimpleString(s) => match s.to_lowercase().as_str() {
                 "ping" => Ok(Self::Ping),
-
                 _ => todo!(),
             },
             _ => todo!(),
diff --git a/src/services/cluster/actors/listening_actor.rs b/src/services/cluster/actors/listening_actor.rs
index 13424c75..cfdb3127 100644
--- a/src/services/cluster/actors/listening_actor.rs
+++ b/src/services/cluster/actors/listening_actor.rs
@@ -70,7 +70,11 @@ impl PeerListeningActor {
                     MasterCommand::Ping => {
                         println!("[INFO] Received ping from master");
                     }
+
                     MasterCommand::Replicate { query: _ } => {}
+                    MasterCommand::Sync(v) => {
+                        println!("[INFO] Received sync from master {:?}", v);
+                    }
                 }
             }
         }
diff --git a/src/services/cluster/actors/replication.rs b/src/services/cluster/actors/replication.rs
index 1732d659..64443b37 100644
--- a/src/services/cluster/actors/replication.rs
+++ b/src/services/cluster/actors/replication.rs
@@ -1,5 +1,7 @@
 use crate::services::config::init::get_env;
 use std::sync::atomic::AtomicBool;
+
+use super::types::PeerAddr;
 pub static IS_MASTER_MODE: AtomicBool = AtomicBool::new(true);
 
 #[derive(Debug, Clone)]
@@ -59,7 +61,8 @@ impl Replication {
         ]
     }
 
-    pub fn master_cluster_bind_addr(&self) -> String {
+    pub fn master_cluster_bind_addr(&self) -> PeerAddr {
         format!("{}:{}", self.master_host.as_ref().unwrap(), self.master_port.unwrap() + 10000)
+            .into()
     }
 }
diff --git a/src/services/cluster/actors/types.rs b/src/services/cluster/actors/types.rs
index ee8fba94..e4660ae9 100644
--- a/src/services/cluster/actors/types.rs
+++ b/src/services/cluster/actors/types.rs
@@ -1,4 +1,4 @@
-use crate::make_smart_pointer;
+use crate::{from_to, make_smart_pointer};
 
 use super::replication::Replication;
 
@@ -11,10 +11,10 @@ pub enum PeerKind {
 
 impl PeerKind {
     pub fn accepted_peer_kind(self_repl_id: &str, other_repl_id: &str) -> Self {
-        if other_repl_id == "?" || self_repl_id == other_repl_id {
-            Self::Replica
-        } else {
-            Self::Peer
+        match other_repl_id {
+            "?" => Self::Replica,
+            id if id == self_repl_id => Self::Replica,
+            _ => Self::Peer,
         }
     }
     pub fn connected_peer_kind(self_repl_info: &Replication, other_repl_id: &str) -> Self {
@@ -31,3 +31,13 @@ impl PeerKind {
 #[derive(Debug, PartialEq, Eq, PartialOrd, Ord, Clone, Hash)]
 pub struct PeerAddr(pub String);
 make_smart_pointer!(PeerAddr, String);
+from_to!(String, PeerAddr);
+
+pub struct PeerAddrs(pub Vec<PeerAddr>);
+make_smart_pointer!(PeerAddrs, Vec<PeerAddr>);
+from_to!(Vec<PeerAddr>, PeerAddrs);
+impl PeerAddrs {
+    pub fn stringify(self) -> String {
+        self.0.into_iter().map(|x| x.0).collect::<Vec<String>>().join(" ")
+    }
+}
diff --git a/src/services/cluster/inbound/stream.rs b/src/services/cluster/inbound/stream.rs
index ee3cff58..39e1eecf 100644
--- a/src/services/cluster/inbound/stream.rs
+++ b/src/services/cluster/inbound/stream.rs
@@ -1,24 +1,34 @@
 use crate::make_smart_pointer;
-
+use crate::services::cluster::actors::command::ClusterCommand;
 use crate::services::cluster::actors::replication::Replication;
 use crate::services::cluster::actors::types::PeerAddr;
+use crate::services::cluster::actors::types::PeerAddrs;
+use crate::services::cluster::actors::types::PeerKind;
 use crate::services::cluster::inbound::request::HandShakeRequest;
 use crate::services::cluster::inbound::request::HandShakeRequestEnum;
 
 use crate::services::interface::{TGetPeerIp, TStream};
 use crate::services::query_io::QueryIO;
+use crate::services::statefuls::cache::manager::CacheManager;
+use crate::services::statefuls::persist::endec::encoder::encoding_processor::EncodingProcessor;
+use anyhow::Context;
 use tokio::net::TcpStream;
 
 // The following is used only when the node is in master mode
-pub(crate) struct InboundStream(pub TcpStream);
+pub(crate) struct InboundStream {
+    pub(crate) stream: TcpStream,
+    pub(crate) repl_info: Replication,
+    pub(crate) inbound_peer_addr: Option<PeerAddr>,
+    pub(crate) inbound_master_replid: Option<String>,
+}
 
-make_smart_pointer!(InboundStream, TcpStream);
+make_smart_pointer!(InboundStream, TcpStream => stream);
 
 impl InboundStream {
-    pub async fn recv_threeway_handshake(
-        &mut self,
-        repl_info: &Replication,
-    ) -> anyhow::Result<(PeerAddr, String)> {
+    pub(crate) fn new(stream: TcpStream, repl_info: Replication) -> Self {
+        Self { stream, repl_info, inbound_peer_addr: None, inbound_master_replid: None }
+    }
+    pub async fn recv_threeway_handshake(&mut self) -> anyhow::Result<()> {
         self.recv_ping().await?;
 
         let port = self.recv_replconf_listening_port().await?;
@@ -27,9 +37,12 @@ impl InboundStream {
         let _capa_val_vec = self.recv_replconf_capa().await?;
 
         // TODO check repl_id is '?' or of mine. If not, consider incoming as peer
-        let (_repl_id, _offset) = self.recv_psync(repl_info).await?;
+        let (repl_id, _offset) = self.recv_psync().await?;
+
+        self.inbound_peer_addr = Some(format!("{}:{}", self.get_peer_ip()?, port).into());
+        self.inbound_master_replid = Some(repl_id);
 
-        Ok((PeerAddr(format!("{}:{}", self.get_peer_ip()?, port)), _repl_id))
+        Ok(())
     }
 
     async fn recv_ping(&mut self) -> anyhow::Result<()> {
@@ -53,13 +66,16 @@ impl InboundStream {
         self.write(QueryIO::SimpleString("OK".to_string())).await?;
         Ok(capa_val_vec)
     }
-    async fn recv_psync(&mut self, repl_info: &Replication) -> anyhow::Result<(String, i64)> {
+    async fn recv_psync(&mut self) -> anyhow::Result<(String, i64)> {
         let mut cmd = self.extract_cmd().await?;
         let (repl_id, offset) = cmd.extract_psync()?;
 
+        let (self_master_replid, self_master_repl_offset) =
+            (self.repl_info.master_replid.clone(), self.repl_info.master_repl_offset);
+
         self.write(QueryIO::SimpleString(format!(
             "FULLRESYNC {} {}",
-            repl_info.master_replid, repl_info.master_repl_offset
+            self_master_replid, self_master_repl_offset
         )))
         .await?;
 
@@ -70,4 +86,52 @@ impl InboundStream {
         let query_io = self.read_value().await?;
         HandShakeRequest::new(query_io)
     }
+
+    pub(crate) async fn disseminate_peers(&mut self, peers: PeerAddrs) -> anyhow::Result<()> {
+        self.write(QueryIO::SimpleString(format!("PEERS {}", peers.stringify()))).await?;
+        Ok(())
+    }
+
+    pub(crate) fn peer_kind(&self) -> anyhow::Result<PeerKind> {
+        Ok(PeerKind::accepted_peer_kind(
+            &self.repl_info.master_replid,
+            self.inbound_master_replid.as_ref().context("No master replid")?,
+        ))
+    }
+
+    pub(crate) fn to_add_peer(self) -> anyhow::Result<ClusterCommand> {
+        Ok(ClusterCommand::AddPeer {
+            peer_kind: self.peer_kind()?,
+            peer_addr: self.inbound_peer_addr.context("No peer addr")?,
+            stream: self.stream,
+        })
+    }
+
+    pub(crate) async fn send_sync_to_inbound_server(
+        &mut self,
+        cache_manager: &'static CacheManager,
+    ) -> anyhow::Result<()> {
+        let (tx, mut rx) = tokio::sync::mpsc::channel(100);
+        // route save caches
+        cache_manager.route_save(tx).await;
+
+        // run encoding processor
+        let mut processor = EncodingProcessor::with_vec(cache_manager.inboxes.len());
+        processor.add_meta().await?;
+        while let Some(cmd) = rx.recv().await {
+            match processor.handle_cmd(cmd).await {
+                Ok(true) => break,
+                Ok(false) => continue,
+                Err(err) => {
+                    panic!("Encoding Error: {:?}", err);
+                }
+            }
+        }
+        // collect dump data from processor
+        let dump = QueryIO::File(processor.into_inner());
+        println!("[INFO] Sent sync to slave {:?}", dump);
+        self.write(dump).await?;
+
+        Ok(())
+    }
 }
diff --git a/src/services/cluster/manager.rs b/src/services/cluster/manager.rs
index c44b18f9..46529196 100644
--- a/src/services/cluster/manager.rs
+++ b/src/services/cluster/manager.rs
@@ -1,16 +1,12 @@
 use super::actors::actor::ClusterActor;
 use super::actors::command::ClusterCommand;
-use super::actors::replication::Replication;
-use super::actors::types::{PeerAddr, PeerKind};
+use super::actors::replication::{Replication, IS_MASTER_MODE};
+use super::actors::types::{PeerAddr, PeerAddrs, PeerKind};
 use crate::make_smart_pointer;
 use crate::services::cluster::inbound::stream::InboundStream;
 use crate::services::cluster::outbound::stream::OutboundStream;
-
-use crate::services::interface::TStream;
-use crate::services::query_io::QueryIO;
-
+use crate::services::statefuls::cache::manager::CacheManager;
 use std::time::Duration;
-use tokio::net::TcpStream;
 use tokio::sync::mpsc::Sender;
 use tokio::time::interval;
 
@@ -40,41 +36,28 @@ impl ClusterManager {
         Self(actor_handler)
     }
 
-    pub(crate) async fn get_peers(&self) -> anyhow::Result<Vec<PeerAddr>> {
+    pub(crate) async fn get_peers(&self) -> anyhow::Result<PeerAddrs> {
         let (tx, rx) = tokio::sync::oneshot::channel();
         self.send(ClusterCommand::GetPeers(tx)).await?;
         let peers = rx.await?;
         Ok(peers)
     }
 
-    pub(crate) async fn accept_peer(&self, mut peer_stream: InboundStream) -> anyhow::Result<()> {
-        let repl_info = self.replication_info().await?;
-
-        let (peer_addr, master_repl_id) = peer_stream.recv_threeway_handshake(&repl_info).await?;
-
-        // TODO Need to decide which point to send file data
-        // TODO At this point, slave stream must write master_replid so that other nodes can tell where it belongs
-
-        self.disseminate_peers(&mut peer_stream).await?;
+    pub(crate) async fn accept_inbound_stream(
+        &self,
+        mut peer_stream: InboundStream,
+        cache_manager: &'static CacheManager,
+    ) -> anyhow::Result<()> {
+        peer_stream.recv_threeway_handshake().await?;
+        peer_stream.disseminate_peers(self.get_peers().await?).await?;
 
-        // TODO At this point again, slave tries to connect to other nodes as peer in the cluster
-        self.send(ClusterCommand::AddPeer {
-            peer_addr,
-            stream: peer_stream.0,
-            peer_kind: PeerKind::accepted_peer_kind(&repl_info.master_replid, &master_repl_id),
-        })
-        .await?;
-        Ok(())
-    }
+        if matches!(peer_stream.peer_kind()?, PeerKind::Replica)
+            && IS_MASTER_MODE.load(std::sync::atomic::Ordering::Acquire)
+        {
+            peer_stream.send_sync_to_inbound_server(cache_manager).await?;
+        }
+        self.send(peer_stream.to_add_peer()?).await?;
 
-    async fn disseminate_peers(&self, stream: &mut TcpStream) -> anyhow::Result<()> {
-        let peers = self.get_peers().await?;
-        stream
-            .write(QueryIO::SimpleString(format!(
-                "PEERS {}",
-                peers.iter().map(|x| x.to_string()).collect::<Vec<_>>().join(" ")
-            )))
-            .await?;
         Ok(())
     }
 
@@ -87,36 +70,28 @@ impl ClusterManager {
     pub(crate) async fn discover_cluster(
         &'static self,
         self_port: u16,
-        connect_to: String,
+        connect_to: PeerAddr,
     ) -> anyhow::Result<()> {
-        let existing_peers = self.get_peers().await?;
-        let repl_info = self.replication_info().await?;
-
         // Base case
-        if existing_peers.contains(&PeerAddr(connect_to.clone())) {
+        let existing_peers = self.get_peers().await?;
+        if existing_peers.contains(&connect_to) {
             return Ok(());
         }
 
-        let mut outbound_stream = OutboundStream(TcpStream::connect(&connect_to).await?);
-
-        let connection_info = outbound_stream.establish_connection(self_port).await?;
-
-        self.send(ClusterCommand::AddPeer {
-            peer_addr: PeerAddr(connect_to),
-            stream: outbound_stream.0,
-            peer_kind: PeerKind::connected_peer_kind(&repl_info, &connection_info.repl_id),
-        })
-        .await?;
-
-        if repl_info.master_replid == "?" {
-            self.send(ClusterCommand::SetReplicationId(connection_info.repl_id.clone())).await?;
-        }
-
-        for peer in connection_info.list_peer_binding_addrs() {
-            // Recursive async calls need indirection because the compiler needs to know the size of the Future at compile time.
-            // async fn returns an anonymous Future type that contains all the state needed to execute the async function. With recursion, this would create an infinitely nested type like:
-            // Future<Future<Future<...>>>
-            // Using Box::pin adds a layer of indirection through the heap, breaking the infinite size issue by making the Future size fixed
+        // Recursive case
+        let (add_peer_cmd, connected_node_info) =
+            OutboundStream::new(connect_to, self.replication_info().await?)
+                .await?
+                .establish_connection(self_port)
+                .await?
+                .set_replication_id(self)
+                .await?
+                .deconstruct()?;
+        self.send(add_peer_cmd).await?;
+
+        // Discover additional peers concurrently
+        for peer in connected_node_info.list_peer_binding_addrs() {
+            // TODO Require investigation. Why does this have to be called at here?
             Box::pin(self.discover_cluster(self_port, peer)).await?;
         }
 
diff --git a/src/services/cluster/outbound/stream.rs b/src/services/cluster/outbound/stream.rs
index 54a79530..140d7907 100644
--- a/src/services/cluster/outbound/stream.rs
+++ b/src/services/cluster/outbound/stream.rs
@@ -1,17 +1,37 @@
+use crate::services::cluster::actors::command::ClusterCommand;
+use crate::services::cluster::actors::replication::Replication;
+use crate::services::cluster::actors::types::{PeerAddr, PeerKind};
+use crate::services::cluster::manager::ClusterManager;
 use crate::services::cluster::outbound::response::ConnectionResponse;
 use crate::services::interface::{TRead, TStream};
 use crate::services::query_io::QueryIO;
 use crate::{make_smart_pointer, write_array};
+use anyhow::Context;
 use tokio::net::TcpStream;
 
 // The following is used only when the node is in slave mode
-pub(crate) struct OutboundStream(pub(crate) TcpStream);
+pub(crate) struct OutboundStream {
+    pub(crate) stream: TcpStream,
+    pub(crate) repl_info: Replication,
+    connected_node_info: Option<ConnectedNodeInfo>,
+    connect_to: PeerAddr,
+}
+make_smart_pointer!(OutboundStream, TcpStream => stream);
+
 impl OutboundStream {
-    pub async fn establish_connection(&mut self, self_port: u16) -> anyhow::Result<ConnectionInfo> {
+    pub(crate) async fn new(connect_to: PeerAddr, repl_info: Replication) -> anyhow::Result<Self> {
+        Ok(OutboundStream {
+            stream: TcpStream::connect(&connect_to.0).await?,
+            repl_info,
+            connected_node_info: None,
+            connect_to,
+        })
+    }
+    pub async fn establish_connection(mut self, self_port: u16) -> anyhow::Result<Self> {
         // Trigger
         self.write(write_array!("PING")).await?;
         let mut ok_count = 0;
-        let mut connection_info = ConnectionInfo::default();
+        let mut connection_info = ConnectedNodeInfo::default();
 
         loop {
             let res = self.read_values().await?;
@@ -34,7 +54,8 @@ impl OutboundStream {
                     ConnectionResponse::PEERS(peer_list) => {
                         println!("[INFO] Received peer list: {:?}", peer_list);
                         connection_info.peer_list = peer_list;
-                        return Ok(connection_info);
+                        self.connected_node_info = Some(connection_info);
+                        return Ok(self);
                     }
                 }
             }
@@ -51,12 +72,40 @@ impl OutboundStream {
             _ => Err(anyhow::anyhow!("Unexpected OK count")),
         }
     }
-}
 
-make_smart_pointer!(OutboundStream, TcpStream);
+    pub(crate) async fn set_replication_id(
+        self,
+        cluster_manager: &ClusterManager,
+    ) -> anyhow::Result<Self> {
+        if self.repl_info.master_replid == "?" {
+            cluster_manager
+                .send(ClusterCommand::SetReplicationId(
+                    self.connected_node_info
+                        .as_ref()
+                        .context("Connected node info not found. Cannot set replication id")?
+                        .repl_id
+                        .clone(),
+                ))
+                .await?;
+        }
+        Ok(self)
+    }
+    pub(crate) fn deconstruct(self) -> anyhow::Result<(ClusterCommand, ConnectedNodeInfo)> {
+        let connection_info = self.connected_node_info.context("Connected node info not found")?;
+
+        Ok((
+            ClusterCommand::AddPeer {
+                peer_kind: PeerKind::connected_peer_kind(&self.repl_info, &connection_info.repl_id),
+                peer_addr: self.connect_to,
+                stream: self.stream,
+            },
+            connection_info,
+        ))
+    }
+}
 
 #[derive(Debug, Default)]
-pub(crate) struct ConnectionInfo {
+pub(crate) struct ConnectedNodeInfo {
     // TODO repl_id here is the master_replid from connected server.
     // TODO Set repl_id if given server's repl_id is "?" otherwise, it means that now it's connected to peer.
     pub(crate) repl_id: String,
@@ -64,16 +113,17 @@ pub(crate) struct ConnectionInfo {
     pub(crate) peer_list: Vec<String>,
 }
 
-impl ConnectionInfo {
-    pub(crate) fn list_peer_binding_addrs(&self) -> Vec<String> {
+impl ConnectedNodeInfo {
+    pub(crate) fn list_peer_binding_addrs(&self) -> Vec<PeerAddr> {
         self.peer_list
             .iter()
             .flat_map(|peer| {
                 if let Some((ip, port)) = peer.rsplit_once(':') {
                     Some(
-                        ip.to_string()
+                        (ip.to_string()
                             + ":"
-                            + (port.parse::<u16>().unwrap() + 10000).to_string().as_str(),
+                            + (port.parse::<u16>().unwrap() + 10000).to_string().as_str())
+                        .into(),
                     )
                 } else {
                     None
diff --git a/src/services/query_io.rs b/src/services/query_io.rs
index b9efbdce..570ff11f 100644
--- a/src/services/query_io.rs
+++ b/src/services/query_io.rs
@@ -11,6 +11,7 @@ pub enum QueryIO {
     Array(Vec<QueryIO>),
     Null,
     Err(String),
+    File(Vec<u8>),
 }
 
 #[macro_export]
@@ -34,6 +35,15 @@ impl QueryIO {
             }
             QueryIO::Null => "$-1\r\n".to_string(),
             QueryIO::Err(e) => format!("-{}\r\n", e),
+            QueryIO::File(f) => {
+                let mut hex_file = String::new();
+                for byte in f {
+                    hex_file.push_str(&format!("{:02x}", byte));
+                }
+                let mut result = format!("${}\r\n", hex_file.len());
+                result.push_str(&hex_file);
+                result
+            }
         }
     }
 
@@ -70,12 +80,18 @@ impl From<String> for QueryIO {
         QueryIO::BulkString(v)
     }
 }
+
+impl From<Vec<u8>> for QueryIO {
+    fn from(v: Vec<u8>) -> Self {
+        QueryIO::File(v)
+    }
+}
 //
 pub fn parse(buffer: BytesMut) -> Result<(QueryIO, usize)> {
     match buffer[0] as char {
         '+' => parse_simple_string(buffer),
         '*' => parse_array(buffer),
-        '$' => parse_bulk_string(buffer),
+        '$' => parse_bulk_string_or_file(buffer),
         _ => Err(anyhow::anyhow!("Not a known value type {:?}", buffer)),
     }
 }
@@ -106,20 +122,27 @@ fn parse_array(buffer: BytesMut) -> Result<(QueryIO, usize)> {
     Ok((QueryIO::Array(bulk_strings), len))
 }
 
-fn parse_bulk_string(buffer: BytesMut) -> Result<(QueryIO, usize)> {
+fn parse_bulk_string_or_file(buffer: BytesMut) -> Result<(QueryIO, usize)> {
     let (line, mut len) =
         read_until_crlf(&buffer[1..]).ok_or(anyhow::anyhow!("Invalid bulk string"))?;
 
     // Adjust `len` to include the initial line and calculate `bulk_str_len`
     len += 1;
-    let bulk_str_len = usize::try_from(ConversionWrapper(line))?;
-
-    // Extract the bulk string from the buffer
-    let bulk_str = &buffer[len..len + bulk_str_len];
-    let bulk_string_value = String::from_utf8(bulk_str.to_vec())?;
-
-    // Return the bulk string value and adjusted length to account for CRLF
-    Ok((QueryIO::BulkString(bulk_string_value), len + bulk_str_len + 2))
+    let content_len = usize::try_from(ConversionWrapper(line))?;
+
+    if let Some((line, _)) = read_until_crlf(&buffer[len..]) {
+        // Extract the bulk string from the buffer
+        let bulk_string_value = String::from_utf8(line.to_vec())?;
+        // Return the bulk string value and adjusted length to account for CRLF
+        Ok((QueryIO::BulkString(bulk_string_value), len + content_len + 2))
+    } else {
+        let file_content = &buffer[len..(len + content_len)];
+        let file = file_content
+            .chunks(2)
+            .map(|chunk| u8::from_str_radix(std::str::from_utf8(chunk).unwrap(), 16))
+            .collect::<Result<Vec<u8>, _>>()?;
+        Ok((QueryIO::File(file), len + content_len))
+    }
 }
 
 fn read_until_crlf(buffer: &[u8]) -> Option<(&[u8], usize)> {
@@ -212,3 +235,17 @@ fn test_parse_array() {
         ])
     );
 }
+
+#[test]
+fn test_parse_file() {
+    // GIVEN
+    let file = QueryIO::File(b"hello".to_vec());
+    let serialized = file.serialize();
+    let buffer = BytesMut::from(serialized.as_str());
+    // WHEN
+    let (value, len) = parse(buffer).unwrap();
+
+    // THEN
+    assert_eq!(len, 15);
+    assert_eq!(value, QueryIO::File(b"hello".to_vec()));
+}
\ No newline at end of file
diff --git a/src/services/statefuls/cache/actor.rs b/src/services/statefuls/cache/actor.rs
index 8bd62c73..67930937 100644
--- a/src/services/statefuls/cache/actor.rs
+++ b/src/services/statefuls/cache/actor.rs
@@ -3,7 +3,7 @@ use super::CacheValue;
 use crate::make_smart_pointer;
 use crate::services::query_io::QueryIO;
 use crate::services::statefuls::cache::ttl::manager::TtlSchedulerInbox;
-use crate::services::statefuls::persist::save_command::SaveCommand;
+use crate::services::statefuls::persist::encoding_command::EncodingCommand;
 use anyhow::Context;
 use anyhow::Result;
 use std::collections::HashMap;
@@ -11,7 +11,7 @@ use tokio::sync::{mpsc, oneshot};
 
 pub enum CacheCommand {
     Set { cache_entry: CacheEntry, ttl_sender: TtlSchedulerInbox },
-    Save { outbox: mpsc::Sender<SaveCommand> },
+    Save { outbox: mpsc::Sender<EncodingCommand> },
     Get { key: String, sender: oneshot::Sender<QueryIO> },
     Keys { pattern: Option<String>, sender: oneshot::Sender<QueryIO> },
     Delete(String),
@@ -63,16 +63,16 @@ impl CacheActor {
                 }
                 CacheCommand::Save { outbox } => {
                     outbox
-                        .send(SaveCommand::LocalShardSize {
+                        .send(EncodingCommand::LocalShardSize {
                             table_size: self.len(),
                             expiry_size: self.keys_with_expiry(),
                         })
                         .await?;
                     for chunk in self.cache.iter().collect::<Vec<_>>().chunks(10) {
-                        outbox.send(SaveCommand::SaveChunk(CacheEntry::new(chunk))).await?;
+                        outbox.send(EncodingCommand::SaveChunk(CacheEntry::new(chunk))).await?;
                     }
                     // finalize the save operation
-                    outbox.send(SaveCommand::StopSentinel).await?;
+                    outbox.send(EncodingCommand::StopSentinel).await?;
                 }
             }
         }
@@ -86,7 +86,7 @@ impl CacheActor {
         self.cache.keys_with_expiry
     }
 
-    fn keys_stream(&self, pattern: Option<String>) -> impl Iterator<Item = QueryIO> + '_ {
+    fn keys_stream(&self, pattern: Option<String>) -> impl Iterator<Item=QueryIO> + '_ {
         self.cache.keys().filter_map(move |k| {
             if pattern.as_ref().map_or(true, |p| k.contains(p)) {
                 Some(QueryIO::BulkString(k.to_string()))
@@ -161,8 +161,8 @@ async fn test_set_and_delete_inc_dec_keys_with_expiry() {
             },
             ttl_sender: ttl_sender.clone(),
         })
-        .await
-        .unwrap();
+            .await
+            .unwrap();
     }
 
     // key0 is expiry key. deleting the following will decrese the number by 1
diff --git a/src/services/statefuls/cache/manager.rs b/src/services/statefuls/cache/manager.rs
index 3e49c788..ed936662 100644
--- a/src/services/statefuls/cache/manager.rs
+++ b/src/services/statefuls/cache/manager.rs
@@ -4,7 +4,7 @@ use crate::services::query_io::QueryIO;
 use crate::services::statefuls::cache::ttl::manager::TtlSchedulerInbox;
 
 use crate::services::statefuls::cache::ttl::actor::TtlActor;
-use crate::services::statefuls::persist::save_command::SaveCommand;
+use crate::services::statefuls::persist::encoding_command::EncodingCommand;
 use crate::services::statefuls::persist::DumpFile;
 use anyhow::Result;
 use std::time::SystemTime;
@@ -54,7 +54,7 @@ impl CacheManager {
         Ok(QueryIO::SimpleString("OK".to_string()))
     }
 
-    pub(crate) async fn route_save(&self, outbox: mpsc::Sender<SaveCommand>) {
+    pub(crate) async fn route_save(&self, outbox: mpsc::Sender<EncodingCommand>) {
         // get all the handlers to cache actors
         for inbox in self.inboxes.iter().map(Clone::clone) {
             let outbox = outbox.clone();
diff --git a/src/services/statefuls/persist/actor.rs b/src/services/statefuls/persist/actor.rs
index 8358249d..5720a5cd 100644
--- a/src/services/statefuls/persist/actor.rs
+++ b/src/services/statefuls/persist/actor.rs
@@ -1,12 +1,12 @@
 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::endec::encoder::encoding_processor::EncodingProcessor;
 use super::DumpFile;
 
 use crate::services::interface::TWriterFactory;
-use crate::services::statefuls::persist::save_command::SaveCommand;
-use tokio::sync::mpsc::{Receiver, Sender};
+use crate::services::statefuls::persist::encoding_command::EncodingCommand;
+use tokio::sync::mpsc::Receiver;
+use tokio::sync::mpsc::Sender;
 
 pub struct Load;
 
@@ -23,15 +23,15 @@ impl PersistActor<Load> {
     }
 }
 
-impl PersistActor<SavingProcessor> {
+impl PersistActor<EncodingProcessor> {
     pub async fn run(
         filepath: String,
         num_of_cache_actors: usize,
-    ) -> anyhow::Result<Sender<SaveCommand>> {
+    ) -> anyhow::Result<Sender<EncodingCommand>> {
         // * Propagate error to caller before sending it to the background
         let file = tokio::fs::File::create_writer(filepath).await?;
 
-        let processor = SavingProcessor::new(file, SaveMeta::new(num_of_cache_actors));
+        let processor = EncodingProcessor::with_file(file, num_of_cache_actors);
 
         let persist_actor = Self { processor };
 
@@ -41,7 +41,7 @@ impl PersistActor<SavingProcessor> {
         Ok(outbox)
     }
 
-    async fn save(mut self, mut inbox: Receiver<SaveCommand>) -> anyhow::Result<()> {
+    async fn save(mut self, mut inbox: Receiver<EncodingCommand>) -> anyhow::Result<()> {
         self.processor.add_meta().await?;
         while let Some(command) = inbox.recv().await {
             match self.processor.handle_cmd(command).await {
diff --git a/src/services/statefuls/persist/save_command.rs b/src/services/statefuls/persist/encoding_command.rs
similarity index 86%
rename from src/services/statefuls/persist/save_command.rs
rename to src/services/statefuls/persist/encoding_command.rs
index 58629ed1..4d1a390f 100644
--- a/src/services/statefuls/persist/save_command.rs
+++ b/src/services/statefuls/persist/encoding_command.rs
@@ -1,6 +1,6 @@
 use crate::services::statefuls::cache::CacheEntry;
 
-pub enum SaveCommand {
+pub enum EncodingCommand {
     LocalShardSize { table_size: usize, expiry_size: usize },
     SaveChunk(Vec<CacheEntry>),
     StopSentinel,
diff --git a/src/services/statefuls/persist/endec/encoder/encoding_processor.rs b/src/services/statefuls/persist/endec/encoder/encoding_processor.rs
index b23ffbc3..3c3e75a4 100644
--- a/src/services/statefuls/persist/endec/encoder/encoding_processor.rs
+++ b/src/services/statefuls/persist/endec/encoder/encoding_processor.rs
@@ -1,23 +1,48 @@
 use crate::services::{interface::TWrite, statefuls::cache::CacheEntry};
 
-use crate::services::statefuls::persist::save_command::SaveCommand;
+use crate::services::statefuls::persist::encoding_command::EncodingCommand;
 use anyhow::Result;
 
+use super::byte_encoder::encode_checksum;
+use super::byte_encoder::encode_database_info;
+use super::byte_encoder::encode_database_table_size;
+use super::byte_encoder::encode_header;
+use super::byte_encoder::encode_metadata;
+use crate::services::error::IoError;
 use std::collections::VecDeque;
+use tokio::io::AsyncWriteExt;
 
-use super::byte_encoder::{
-    encode_checksum, encode_database_info, encode_database_table_size, encode_header,
-    encode_metadata,
-};
+pub enum EncodingTarget {
+    File(tokio::fs::File),
+    InMemory(Vec<u8>),
+}
+
+impl EncodingTarget {
+    pub async fn write(&mut self, buf: &[u8]) -> Result<(), IoError> {
+        match self {
+            EncodingTarget::File(f) => {
+                f.write_all(buf).await
+                    .map_err(|e| e.kind().into())
+            }
+            EncodingTarget::InMemory(v) => {
+                Ok(v.extend_from_slice(buf))
+            }
+        }
+    }
+}
 
-pub struct SavingProcessor {
-    pub(super) writer: tokio::fs::File,
+pub struct EncodingProcessor {
+    pub(super) target: EncodingTarget,
     pub(super) meta: SaveMeta,
 }
 
-impl SavingProcessor {
-    pub fn new(file: tokio::fs::File, meta: SaveMeta) -> Self {
-        Self { writer: file, meta }
+impl EncodingProcessor {
+    pub fn with_file(file: tokio::fs::File, num_of_cache_actor: usize) -> Self {
+        Self { target: EncodingTarget::File(file), meta: SaveMeta::new(num_of_cache_actor) }
+    }
+
+    pub fn with_vec(num_of_cache_actor: usize) -> Self {
+        Self { target: EncodingTarget::InMemory(Vec::new()), meta: SaveMeta::new(num_of_cache_actor) }
     }
 
     pub async fn add_meta(&mut self) -> Result<()> {
@@ -26,17 +51,17 @@ impl SavingProcessor {
             encode_metadata(Vec::from([("redis-ver", "6.0.16")]))?,
             encode_database_info(0)?,
         ];
-        self.writer.write(&meta.concat()).await?;
+        self.target.write(&meta.concat()).await?;
         Ok(())
     }
-    pub async fn handle_cmd(&mut self, cmd: SaveCommand) -> Result<bool> {
+    pub async fn handle_cmd(&mut self, cmd: EncodingCommand) -> Result<bool> {
         match cmd {
-            SaveCommand::LocalShardSize { table_size, expiry_size } => {
+            EncodingCommand::LocalShardSize { table_size, expiry_size } => {
                 self.meta.total_key_value_table_size += table_size;
                 self.meta.total_expires_table_size += expiry_size;
                 self.meta.num_of_saved_table_size_actor -= 1;
                 if self.meta.num_of_saved_table_size_actor == 0 {
-                    self.writer
+                    self.target
                         .write(&encode_database_table_size(
                             self.meta.total_key_value_table_size,
                             self.meta.total_expires_table_size,
@@ -44,18 +69,18 @@ impl SavingProcessor {
                         .await?;
                 }
             }
-            SaveCommand::SaveChunk(chunk) => {
+            EncodingCommand::SaveChunk(chunk) => {
                 self.meta.chunk_queue.push_back(chunk);
                 if self.meta.num_of_saved_table_size_actor == 0 {
                     self.encode_chunk_queue().await?;
                 }
             }
-            SaveCommand::StopSentinel => {
+            EncodingCommand::StopSentinel => {
                 self.meta.num_of_cache_actors -= 1;
                 if self.meta.num_of_cache_actors == 0 {
                     self.encode_chunk_queue().await?;
                     let checksum = encode_checksum(&[0; 8])?;
-                    self.writer.write(&checksum).await?;
+                    self.target.write(&checksum).await?;
                     return Ok(true);
                 }
             }
@@ -67,11 +92,19 @@ impl SavingProcessor {
         while let Some(chunk) = self.meta.chunk_queue.pop_front() {
             for kvs in chunk {
                 let encoded_chunk = kvs.encode_with_key()?;
-                self.writer.write(&encoded_chunk).await?;
+                self.target.write(&encoded_chunk).await?;
             }
         }
         Ok(())
     }
+
+    pub fn into_inner(self) -> Vec<u8> {
+        if let EncodingTarget::InMemory(v) = self.target {
+            v
+        } else {
+            panic!("cannot return inner for non InMemory type target")
+        }
+    }
 }
 
 pub struct SaveMeta {
@@ -92,4 +125,4 @@ impl SaveMeta {
             num_of_cache_actors,
         }
     }
-}
+}
\ No newline at end of file
diff --git a/src/services/statefuls/persist/mod.rs b/src/services/statefuls/persist/mod.rs
index f532f382..c5aefb0f 100644
--- a/src/services/statefuls/persist/mod.rs
+++ b/src/services/statefuls/persist/mod.rs
@@ -1,6 +1,6 @@
 pub mod actor;
 pub mod endec;
-pub mod save_command;
+pub mod encoding_command;
 use super::cache::CacheEntry;
 use std::collections::HashMap;
 


Test patch:
diff --git a/tests/common.rs b/tests/common.rs
index da052bf5..7b1605e4 100644
--- a/tests/common.rs
+++ b/tests/common.rs
@@ -73,18 +73,41 @@ pub fn run_server_process(port: u16, replicaof: Option<String>) -> TestProcessCh
 
 pub fn wait_for_message<T: Read>(read: &mut T, target: &str, target_count: usize) {
     let mut buf = BufReader::new(read).lines();
-    let mut cnt = 0;
+    let mut cnt = 1;
 
     while let Some(Ok(line)) = buf.next() {
-        if line == target {
-            cnt += 1;
+        if line.starts_with(target) {
             if cnt == target_count {
                 break;
+            } else {
+                cnt += 1;
             }
         }
     }
 }
 
+pub fn wait_for_and_get_message<T: Read>(
+    read: &mut T,
+    target: &str,
+    target_count: usize,
+) -> String {
+    let mut buf = BufReader::new(read).lines();
+    let mut cnt = 1;
+    let mut message = String::new();
+
+    while let Some(Ok(line)) = buf.next() {
+        if line.starts_with(target) {
+            if cnt == target_count {
+                message = line;
+                break;
+            } else {
+                cnt += 1;
+            }
+        }
+    }
+    message
+}
+
 pub fn array(arr: Vec<&str>) -> String {
     QueryIO::Array(arr.iter().map(|s| QueryIO::BulkString(s.to_string())).collect()).serialize()
 }
diff --git a/tests/test_load_sync_dump.rs b/tests/test_load_sync_dump.rs
new file mode 100644
index 00000000..964665a8
--- /dev/null
+++ b/tests/test_load_sync_dump.rs
@@ -0,0 +1,24 @@
+use crate::common::{spawn_server_as_slave, spawn_server_process, wait_for_and_get_message};
+
+mod common;
+
+#[tokio::test]
+async fn test_send_sync_and_receive_sync() {
+    let mut master = spawn_server_process();
+    let mut replica = spawn_server_as_slave(&master);
+    let mut stdout = master.stdout.take().unwrap();
+
+    let master_message = "[INFO] Sent sync to slave ";
+    let message = wait_for_and_get_message(&mut stdout, master_message, 1);
+
+    let file: String = message.split_inclusive(master_message).skip(1).collect();
+
+    let mut repl_stdout = replica.stdout.take().unwrap();
+
+    let slave_message = "[INFO] Received sync from master ";
+    let message = wait_for_and_get_message(&mut repl_stdout, slave_message, 1);
+
+    let repl_file: String = message.split_inclusive(slave_message).skip(1).collect();
+
+    assert_eq!(file, repl_file);
+}