Repository: Migorithm/duva
Pull request #211 (2025-02-04 15:41:27)

feat: generate-replication-id
Summary:

- **Replid Generator**: A new generator for replication IDs has been added, allowing for the creation of unique identifiers for replication processes.
- **Saving Mechanism**: A mechanism for saving the generated replid along with its corresponding offset has been implemented, enhancing the tracking and management of replication states.
- **RDB Integration**: The replid is now being replaced from the RDB (Redis Database), ensuring that the replication process is aligned with the current state of the database.

Added `rand` dependency


Patch:
diff --git a/Cargo.lock b/Cargo.lock
index 370f61b1..1be5eea3 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -1,6 +1,6 @@
 # This file is automatically @generated by Cargo.
 # It is not intended for manual editing.
-version = 3
+version = 4
 
 [[package]]
 name = "addr2line"
@@ -50,6 +50,12 @@ version = "2.6.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "b048fb63fd8b5923fc5aa7b340d8e156aec7ec02f0c78fa8a6ddc2613f6f71de"
 
+[[package]]
+name = "byteorder"
+version = "1.5.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "1fd0f2584146f6f2ef48085050886acf353beff7305ebd1ae69500e27c67f64b"
+
 [[package]]
 name = "bytes"
 version = "1.6.0"
@@ -74,10 +80,23 @@ version = "0.1.0"
 dependencies = [
  "anyhow",
  "bytes",
+ "rand",
  "thiserror",
  "tokio",
 ]
 
+[[package]]
+name = "getrandom"
+version = "0.3.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "43a49c392881ce6d5c3b8cb70f98717b7c07aabbdff06687b9030dbfbe2725f8"
+dependencies = [
+ "cfg-if",
+ "libc",
+ "wasi 0.13.3+wasi-0.2.2",
+ "windows-targets 0.52.6",
+]
+
 [[package]]
 name = "gimli"
 version = "0.29.0"
@@ -128,7 +147,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "a4a650543ca06a924e8b371db273b2756685faae30f8487da1b56505a8f78b0c"
 dependencies = [
  "libc",
- "wasi",
+ "wasi 0.11.0+wasi-snapshot-preview1",
  "windows-sys 0.48.0",
 ]
 
@@ -180,6 +199,15 @@ version = "0.2.14"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "bda66fc9667c18cb2758a2ac84d1167245054bcf85d5d1aaa6923f45801bdd02"
 
+[[package]]
+name = "ppv-lite86"
+version = "0.2.20"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "77957b295656769bb8ad2b6a6b09d897d94f05c41b069aede1fcdaa675eaea04"
+dependencies = [
+ "zerocopy 0.7.35",
+]
+
 [[package]]
 name = "proc-macro2"
 version = "1.0.86"
@@ -198,6 +226,37 @@ dependencies = [
  "proc-macro2",
 ]
 
+[[package]]
+name = "rand"
+version = "0.9.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "3779b94aeb87e8bd4e834cee3650289ee9e0d5677f976ecdb6d219e5f4f6cd94"
+dependencies = [
+ "rand_chacha",
+ "rand_core",
+ "zerocopy 0.8.14",
+]
+
+[[package]]
+name = "rand_chacha"
+version = "0.9.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "d3022b5f1df60f26e1ffddd6c66e8aa15de382ae63b3a0c1bfc0e4d3e3f325cb"
+dependencies = [
+ "ppv-lite86",
+ "rand_core",
+]
+
+[[package]]
+name = "rand_core"
+version = "0.9.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "b08f3c9802962f7e1b25113931d94f43ed9725bebc59db9d0c3e9a23b67e15ff"
+dependencies = [
+ "getrandom",
+ "zerocopy 0.8.14",
+]
+
 [[package]]
 name = "redox_syscall"
 version = "0.5.2"
@@ -317,6 +376,15 @@ version = "0.11.0+wasi-snapshot-preview1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "9c8d87e72b64a3b4db28d11ce29237c246188f4f51057d65a7eab63b7987e423"
 
+[[package]]
+name = "wasi"
+version = "0.13.3+wasi-0.2.2"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "26816d2e1a4a36a2940b96c5296ce403917633dff8f3440e9b236ed6f6bacad2"
+dependencies = [
+ "wit-bindgen-rt",
+]
+
 [[package]]
 name = "windows-sys"
 version = "0.48.0"
@@ -455,3 +523,53 @@ name = "windows_x86_64_msvc"
 version = "0.52.6"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "589f6da84c646204747d1270a2a5661ea66ed1cced2631d546fdfb155959f9ec"
+
+[[package]]
+name = "wit-bindgen-rt"
+version = "0.33.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "3268f3d866458b787f390cf61f4bbb563b922d091359f9608842999eaee3943c"
+dependencies = [
+ "bitflags",
+]
+
+[[package]]
+name = "zerocopy"
+version = "0.7.35"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "1b9b4fd18abc82b8136838da5d50bae7bdea537c574d8dc1a34ed098d6c166f0"
+dependencies = [
+ "byteorder",
+ "zerocopy-derive 0.7.35",
+]
+
+[[package]]
+name = "zerocopy"
+version = "0.8.14"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "a367f292d93d4eab890745e75a778da40909cab4d6ff8173693812f79c4a2468"
+dependencies = [
+ "zerocopy-derive 0.8.14",
+]
+
+[[package]]
+name = "zerocopy-derive"
+version = "0.7.35"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "fa4f8080344d4671fb4e831a13ad1e68092748387dfc4f55e356242fae12ce3e"
+dependencies = [
+ "proc-macro2",
+ "quote",
+ "syn",
+]
+
+[[package]]
+name = "zerocopy-derive"
+version = "0.8.14"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "d3931cb58c62c13adec22e38686b559c86a30565e16ad6e8510a337cedc611e1"
+dependencies = [
+ "proc-macro2",
+ "quote",
+ "syn",
+]
diff --git a/Cargo.toml b/Cargo.toml
index a76ff52d..6bfb8ee9 100644
--- a/Cargo.toml
+++ b/Cargo.toml
@@ -11,4 +11,5 @@ edition = "2021"
 anyhow = "1.0.59"                                   # error handling
 bytes = "1.3.0"                                     # helps manage buffers
 thiserror = "1.0.32"                                # error handling
-tokio = { version = "1.23.0", features = ["full"] } # async networking
+tokio = { version = "1.23.0", features = ["full"] }
+rand = "0.9.0" # async networking
diff --git a/src/lib.rs b/src/lib.rs
index 2a24c667..d36d632d 100644
--- a/src/lib.rs
+++ b/src/lib.rs
@@ -3,6 +3,7 @@ pub mod macros;
 pub mod services;
 use anyhow::Result;
 use services::client::manager::ClientManager;
+use services::cluster::command::cluster_command::ClusterCommand;
 use services::cluster::inbound::stream::InboundStream;
 use services::cluster::manager::ClusterManager;
 use services::cluster::replication::replication::IS_MASTER_MODE;
@@ -12,7 +13,6 @@ use services::error::IoError;
 use services::statefuls::cache::manager::CacheManager;
 use services::statefuls::cache::ttl::manager::TtlSchedulerInbox;
 use services::statefuls::persist::actor::PersistActor;
-
 use std::sync::atomic::Ordering;
 use std::thread::sleep;
 use std::time::Duration;
@@ -64,23 +64,19 @@ impl StartUpFacade {
     }
 
     pub async fn run(&mut self, startup_notifier: impl TNotifyStartUp) -> Result<()> {
-        if let Some(filepath) = self.config_manager.try_filepath().await? {
-            let dump = PersistActor::dump(filepath).await?;
-            self.cache_manager
-                .dump_cache(dump, self.ttl_inbox.clone(), self.config_manager.startup_time)
-                .await?;
-        }
-
         tokio::spawn(Self::start_accepting_peer_connections(
             self.config_manager.peer_bind_addr(),
             self.cluster_manager,
             self.cache_manager,
         ));
 
-        tokio::spawn(async move {
-            tokio::time::sleep(Duration::from_secs(1)).await;
-            startup_notifier.notify_startup()
-        });
+        tokio::spawn(Self::initialize_with_dump(
+            self.config_manager.clone(),
+            self.cache_manager,
+            self.ttl_inbox.clone(),
+            self.cluster_manager,
+            startup_notifier,
+        ));
 
         self.start_mode_specific_connection_handling().await
     }
@@ -165,6 +161,27 @@ impl StartUpFacade {
     fn cluster_mode(&mut self) -> bool {
         *self.mode_change_watcher.borrow_and_update()
     }
+
+    async fn initialize_with_dump(
+        config_manager: ConfigManager,
+        cache_manager: &'static CacheManager,
+        ttl_inbox: TtlSchedulerInbox,
+        cluster_manager: &'static ClusterManager,
+        startup_notifier: impl TNotifyStartUp,
+    ) -> Result<()> {
+        if let Some(filepath) = config_manager.try_filepath().await? {
+            let dump = PersistActor::dump(filepath).await?;
+            if let Some((repl_id, offset)) = dump.extract_replication_info() {
+                //  TODO reconnect! - echo
+                cluster_manager
+                    .send(ClusterCommand::SetReplicationInfo { master_repl_id: repl_id, offset })
+                    .await?;
+            };
+            cache_manager.dump_cache(dump, ttl_inbox.clone(), config_manager.startup_time).await?;
+        }
+        startup_notifier.notify_startup();
+        Ok(())
+    }
 }
 
 pub trait TNotifyStartUp: Send + 'static {
diff --git a/src/services/client/manager.rs b/src/services/client/manager.rs
index 1bc3a145..a0af3744 100644
--- a/src/services/client/manager.rs
+++ b/src/services/client/manager.rs
@@ -10,7 +10,9 @@ use crate::services::statefuls::cache::manager::CacheManager;
 use crate::services::statefuls::cache::ttl::manager::TtlSchedulerInbox;
 use crate::services::statefuls::cache::CacheEntry;
 use crate::services::statefuls::persist::actor::PersistActor;
-use crate::services::statefuls::persist::endec::encoder::encoding_processor::EncodingProcessor;
+use crate::services::statefuls::persist::endec::encoder::encoding_processor::{
+    EncodingProcessor, SaveMeta,
+};
 use tokio::net::{TcpListener, TcpStream};
 use tokio::select;
 
@@ -48,9 +50,15 @@ impl ClientManager {
             }
             ClientRequest::Save => {
                 // spawn save actor
+                let replication = self.cluster_manager.replication_info().await?;
+                let save_meta = SaveMeta::new(
+                    self.cache_manager.inboxes.len(),
+                    replication.master_replid,
+                    replication.master_repl_offset.to_string(),
+                );
                 let outbox = PersistActor::<EncodingProcessor>::run(
                     self.config_manager.get_filepath().await?,
-                    self.cache_manager.inboxes.len(),
+                    save_meta,
                 )
                 .await?;
 
diff --git a/src/services/cluster/actors/listening_actor.rs b/src/services/cluster/actors/listening_actor.rs
index 7e074501..36537ef1 100644
--- a/src/services/cluster/actors/listening_actor.rs
+++ b/src/services/cluster/actors/listening_actor.rs
@@ -14,7 +14,6 @@ use tokio::select;
 use tokio::sync::mpsc::Sender;
 use tokio::task::JoinHandle;
 
-
 pub(crate) struct PeerListeningActor {
     pub(crate) read_connected: ReadConnected,
     pub(crate) cluster_handler: Sender<ClusterCommand>,
diff --git a/src/services/cluster/command/cluster_command.rs b/src/services/cluster/command/cluster_command.rs
index abadfcaf..5614cfe9 100644
--- a/src/services/cluster/command/cluster_command.rs
+++ b/src/services/cluster/command/cluster_command.rs
@@ -22,4 +22,3 @@ pub struct AddPeer {
     pub(crate) stream: TcpStream,
     pub(crate) peer_kind: PeerKind,
 }
-
diff --git a/src/services/cluster/inbound/stream.rs b/src/services/cluster/inbound/stream.rs
index cb563301..452e8390 100644
--- a/src/services/cluster/inbound/stream.rs
+++ b/src/services/cluster/inbound/stream.rs
@@ -11,7 +11,9 @@ use crate::services::interface::TRead;
 use crate::services::interface::TWrite;
 use crate::services::query_io::QueryIO;
 use crate::services::statefuls::cache::manager::CacheManager;
-use crate::services::statefuls::persist::endec::encoder::encoding_processor::EncodingProcessor;
+use crate::services::statefuls::persist::endec::encoder::encoding_processor::{
+    EncodingProcessor, SaveMeta,
+};
 use anyhow::Context;
 use bytes::Bytes;
 use tokio::net::TcpStream;
@@ -121,8 +123,13 @@ impl InboundStream {
         cache_manager.route_save(tx).await;
 
         // run encoding processor
-        let mut processor = EncodingProcessor::with_vec(cache_manager.inboxes.len());
-        processor.add_meta().await?;
+        let save_meta = SaveMeta::new(
+            cache_manager.inboxes.len(),
+            self.repl_info.master_replid.clone(),
+            self.repl_info.master_repl_offset.to_string(),
+        );
+        let mut processor = EncodingProcessor::with_vec(save_meta);
+        processor.encode_meta().await?;
         while let Some(cmd) = rx.recv().await {
             match processor.handle_cmd(cmd).await {
                 Ok(true) => break,
diff --git a/src/services/cluster/replication/mod.rs b/src/services/cluster/replication/mod.rs
index dc7c1b68..fa866f57 100644
--- a/src/services/cluster/replication/mod.rs
+++ b/src/services/cluster/replication/mod.rs
@@ -1,1 +1,2 @@
-pub(crate) mod replication;
\ No newline at end of file
+pub(crate) mod replication;
+pub(crate) mod replid_generator;
\ No newline at end of file
diff --git a/src/services/cluster/replication/replication.rs b/src/services/cluster/replication/replication.rs
index 003e8710..6dfce6b3 100644
--- a/src/services/cluster/replication/replication.rs
+++ b/src/services/cluster/replication/replication.rs
@@ -2,6 +2,8 @@ use crate::services::config::init::get_env;
 use std::sync::atomic::AtomicBool;
 
 use crate::services::cluster::peer::identifier::PeerIdentifier;
+use crate::services::cluster::replication::replid_generator::generate_replid;
+
 pub static IS_MASTER_MODE: AtomicBool = AtomicBool::new(true);
 
 #[derive(Debug, Clone)]
@@ -36,7 +38,7 @@ impl Default for Replication {
 impl Replication {
     pub fn new(replicaof: Option<(String, String)>, self_host: &str, self_port: u16) -> Self {
         let master_replid = if replicaof.is_none() {
-            "8371b4fb1155b71f4a04d3e1bc3e18c4a990aeeb".to_string()
+            generate_replid()
         } else {
             "?".to_string()
         };
diff --git a/src/services/cluster/replication/replid_generator.rs b/src/services/cluster/replication/replid_generator.rs
new file mode 100644
index 00000000..57156f46
--- /dev/null
+++ b/src/services/cluster/replication/replid_generator.rs
@@ -0,0 +1,36 @@
+use rand::rngs::StdRng;
+use rand::{Rng, SeedableRng};
+use std::process;
+use std::time::{SystemTime, UNIX_EPOCH};
+
+const CHARSET: &[u8] = b"abcdefghijklmnopqrstuvwxyz0123456789";
+const LENGTH: usize = 40;
+
+pub(crate) fn generate_replid() -> String {
+    let time = SystemTime::now()
+        .duration_since(UNIX_EPOCH)
+        .expect("Time went backwards")
+        .as_nanos() as u64;
+
+    let pid = process::id() as u64;
+
+    // each entropy source occupies distinct bit regions, reducing the chance of collisions.
+    let seed = (pid << 32) ^ time;
+
+    let mut rng = StdRng::seed_from_u64(seed);
+    let random_string: String = (0..LENGTH)
+        .map(|_| {
+            let idx = rng.random_range(0..CHARSET.len());
+            CHARSET[idx] as char
+        })
+        .collect();
+
+    random_string
+}
+
+#[test]
+fn test_generate_replid() {
+    let replid = generate_replid();
+    assert_eq!(replid.len(), 40);
+    println!("{}", replid);
+}
\ No newline at end of file
diff --git a/src/services/config/manager.rs b/src/services/config/manager.rs
index f511aa0a..1d391eb2 100644
--- a/src/services/config/manager.rs
+++ b/src/services/config/manager.rs
@@ -6,7 +6,6 @@ use super::ConfigResource;
 use super::ConfigResponse;
 use crate::services::config::init::get_env;
 use std::time::SystemTime;
-
 use tokio::fs::try_exists;
 use tokio::sync::mpsc::Sender;
 use tokio::sync::oneshot;
diff --git a/src/services/statefuls/persist/actor.rs b/src/services/statefuls/persist/actor.rs
index 5720a5cd..cdd37d73 100644
--- a/src/services/statefuls/persist/actor.rs
+++ b/src/services/statefuls/persist/actor.rs
@@ -1,6 +1,6 @@
 use super::endec::decoder::byte_decoder::BytesDecoder;
 use super::endec::decoder::states::DecoderInit;
-use super::endec::encoder::encoding_processor::EncodingProcessor;
+use super::endec::encoder::encoding_processor::{EncodingProcessor, SaveMeta};
 use super::DumpFile;
 
 use crate::services::interface::TWriterFactory;
@@ -26,12 +26,12 @@ impl PersistActor<Load> {
 impl PersistActor<EncodingProcessor> {
     pub async fn run(
         filepath: String,
-        num_of_cache_actors: usize,
+        save_meta: SaveMeta,
     ) -> 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 = EncodingProcessor::with_file(file, num_of_cache_actors);
+        let processor = EncodingProcessor::with_file(file, save_meta);
 
         let persist_actor = Self { processor };
 
@@ -42,7 +42,7 @@ impl PersistActor<EncodingProcessor> {
     }
 
     async fn save(mut self, mut inbox: Receiver<EncodingCommand>) -> anyhow::Result<()> {
-        self.processor.add_meta().await?;
+        self.processor.encode_meta().await?;
         while let Some(command) = inbox.recv().await {
             match self.processor.handle_cmd(command).await {
                 Ok(should_break) => {
diff --git a/src/services/statefuls/persist/endec/decoder/byte_decoder.rs b/src/services/statefuls/persist/endec/decoder/byte_decoder.rs
index 85216e94..ba3191ad 100644
--- a/src/services/statefuls/persist/endec/decoder/byte_decoder.rs
+++ b/src/services/statefuls/persist/endec/decoder/byte_decoder.rs
@@ -4,12 +4,11 @@ use crate::services::statefuls::persist::endec::{
     EXPIRY_TIME_IN_MILLISECONDS_INDICATOR, EXPIRY_TIME_IN_SECONDS_INDICATOR, HEADER_MAGIC_STRING,
     METADATA_SECTION_INDICATOR, STRING_VALUE_TYPE_INDICATOR,
 };
-use crate::services::statefuls::persist::{DatabaseSection, DatabaseSectionBuilder, DumpFile};
-
+use crate::services::statefuls::persist::{
+    DatabaseSection, DatabaseSectionBuilder, DumpFile, DumpMetadata,
+};
 use anyhow::{Context, Result};
-
 use std::{
-    collections::HashMap,
     ops::{Deref, DerefMut},
     time::SystemTime,
 };
@@ -150,12 +149,23 @@ impl<'a> BytesDecoder<'a, DecoderInit> {
 
 impl<'a> BytesDecoder<'a, HeaderReady> {
     pub fn load_metadata(mut self) -> Result<BytesDecoder<'a, MetadataReady>> {
-        let mut metadata = HashMap::new();
+        let mut metadata = DumpMetadata::default();
         while self.check_indicator(METADATA_SECTION_INDICATOR) {
             let (key, value) = self
                 .try_extract_metadata_key_value()
                 .context("metadata loading: key value extraction failed")?;
-            metadata.insert(key, value);
+
+            match key.as_str() {
+                "repl-id" => metadata.repl_id = Some(value),
+                "repl-offset" => {
+                    metadata.repl_offset = Some(value.parse().context("repl-offset parse fail")?)
+                }
+
+                var => {
+                    println!("Unknown metadata key: {}", var);
+                    panic!()
+                }
+            }
         }
         Ok(BytesDecoder {
             data: self.data,
@@ -523,37 +533,6 @@ fn test_header_loading_data_length_error() {
     assert!(result.is_err());
 }
 
-#[test]
-fn test_metadata_loading() {
-    static DATA: [u8; 9] = [0xFA, 0x03, 0x61, 0x62, 0x63, 0x03, 0x64, 0x65, 0x66];
-
-    let bytes_handler = BytesDecoder::<HeaderReady> {
-        data: (&DATA as &'static [u8]),
-        state: HeaderReady("REDIS0001".to_string()),
-    };
-
-    let metadata = bytes_handler.load_metadata().unwrap();
-
-    assert_eq!(metadata.state.metadata.get("abc"), Some(&"def".to_string()));
-    assert_eq!(metadata.state.metadata.get("ghi"), None);
-    assert_eq!(metadata.state.header, "REDIS0001");
-}
-
-#[test]
-fn test_metadata_loading_multiple() {
-    let data = vec![
-        0xFA, 0x03, 0x61, 0x62, 0x63, 0x03, 0x64, 0x65, 0x66, 0xFA, 0x03, 0x67, 0x68, 0x69, 0x03,
-        0x6A, 0x6B, 0x6C,
-    ];
-    let bytes_handler =
-        BytesDecoder::<HeaderReady> { data: data.as_slice().into(), state: Default::default() };
-
-    let metadata = bytes_handler.load_metadata().unwrap();
-
-    assert_eq!(metadata.state.metadata.get("abc"), Some(&"def".to_string()));
-    assert_eq!(metadata.state.metadata.get("ghi"), Some(&"jkl".to_string()));
-}
-
 #[test]
 fn test_metadata_loading_no_metadata() {
     let data = vec![
@@ -564,7 +543,7 @@ fn test_metadata_loading_no_metadata() {
         BytesDecoder::<HeaderReady> { data: data.as_slice().into(), state: Default::default() };
 
     let metadata = bytes_handler.load_metadata().unwrap();
-    assert_eq!(metadata.state.metadata, HashMap::new());
+    assert_eq!(metadata.state.metadata, DumpMetadata::default());
 }
 
 #[test]
@@ -589,15 +568,25 @@ fn test_database_loading() {
 // ! Most important test for the BytesEndec implementation in decoding path.
 #[test]
 fn test_loading_all() {
+    const M_KEY: u8 = 0xFA;
+
+    const R_ID_SIZE: u8 = 0x28;
+    const D_KEY: u8 = 0xFE;
+
     let data = vec![
         // Header
         0x52, 0x45, 0x44, 0x49, 0x53, 0x30, 0x30, 0x31, 0x31, // Metadata
-        0xFA, 0x09, 0x72, 0x65, 0x64, 0x69, 0x73, 0x2D, 0x76, 0x65, 0x72, 0x05, 0x37, 0x2E, 0x32,
-        0x2E, 0x36, 0xFA, 0x0A, 0x72, 0x65, 0x64, 0x69, 0x73, 0x2D, 0x62, 0x69, 0x74, 0x73, 0xC0,
-        0x40, 0xFA, 0x05, 0x63, 0x74, 0x69, 0x6D, 0x65, 0xC2, 0xEA, 0x17, 0x3E, 0x67, 0xFA, 0x08,
-        0x75, 0x73, 0x65, 0x64, 0x2D, 0x6D, 0x65, 0x6D, 0xC2, 0x30, 0xD1, 0x11, 0x00, 0xFA, 0x08,
-        0x61, 0x6F, 0x66, 0x2D, 0x62, 0x61, 0x73, 0x65, 0xC0, 0x00, // Database
-        0xFE, 0x00, 0xFB, 0x02, 0x00, 0x00, 0x04, 0x66, 0x6F, 0x6F, 0x32, 0x04, 0x62, 0x61, 0x72,
+        //* repl-id
+        M_KEY, 0x07, //size of key
+        0x72, 0x65, 0x70, 0x6c, 0x2d, 0x69, 0x64, R_ID_SIZE, // size of value(hex 28 = 40)
+        0x34, 0x32, 0x30, 0x64, 0x64, 0x37, 0x65, 0x33, 0x32, 0x34, 0x63, 0x33, 0x61, 0x36, 0x33,
+        0x37, 0x31, 0x62, 0x31, 0x30, 0x33, 0x31, 0x32, 0x39, 0x63, 0x65, 0x62, 0x65, 0x36, 0x65,
+        0x32, 0x35, 0x61, 0x32, 0x37, 0x30, 0x66, 0x39, 0x66, 0x64, //*
+        M_KEY, 0x0B, //size of key - repl-offset
+        0x72, 0x65, 0x70, 0x6C, 0x2D, 0x6F, 0x66, 0x66, 0x73, 0x65, 0x74,
+        0xC2, //size of offset
+        0xA1, 0xC3, 0x83, 0x00, // Database
+        D_KEY, 0x00, 0xFB, 0x02, 0x00, 0x00, 0x04, 0x66, 0x6F, 0x6F, 0x32, 0x04, 0x62, 0x61, 0x72,
         0x32, 0x00, 0x03, 0x66, 0x6F, 0x6F, 0x03, 0x62, 0x61, 0x72, 0xFF, 0x60, 0x82, 0x9C, 0xF8,
         0xFB, 0x2E, 0x7F, 0xEB,
     ];
@@ -627,4 +616,6 @@ fn test_loading_all() {
     }
 
     assert_eq!(rdb_file.checksum, vec![0x60, 0x82, 0x9C, 0xF8, 0xFB, 0x2E, 0x7F, 0xEB]);
+    assert_eq!(rdb_file.metadata.repl_id.unwrap(), "420dd7e324c3a6371b103129cebe6e25a270f9fd");
+    assert_eq!(rdb_file.metadata.repl_offset.unwrap(), 8635297);
 }
diff --git a/src/services/statefuls/persist/endec/decoder/states.rs b/src/services/statefuls/persist/endec/decoder/states.rs
index 68cc6678..aa3f1028 100644
--- a/src/services/statefuls/persist/endec/decoder/states.rs
+++ b/src/services/statefuls/persist/endec/decoder/states.rs
@@ -1,5 +1,7 @@
 use std::collections::HashMap;
 
+use crate::services::statefuls::persist::DumpMetadata;
+
 #[derive(Default)]
 pub struct DecoderInit;
 
@@ -7,6 +9,6 @@ pub struct DecoderInit;
 pub struct HeaderReady(pub(crate) String);
 #[derive(Default)]
 pub struct MetadataReady {
-    pub(crate) metadata: HashMap<String, String>,
+    pub(crate) metadata: DumpMetadata,
     pub(crate) header: String,
 }
diff --git a/src/services/statefuls/persist/endec/encoder/encoding_processor.rs b/src/services/statefuls/persist/endec/encoder/encoding_processor.rs
index 3b833320..1113e761 100644
--- a/src/services/statefuls/persist/endec/encoder/encoding_processor.rs
+++ b/src/services/statefuls/persist/endec/encoder/encoding_processor.rs
@@ -32,23 +32,22 @@ pub struct EncodingProcessor {
 }
 
 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_file(file: tokio::fs::File, meta: SaveMeta) -> Self {
+        Self { target: EncodingTarget::File(file), meta }
     }
 
-    pub fn with_vec(num_of_cache_actor: usize) -> Self {
-        Self {
-            target: EncodingTarget::InMemory(Vec::new()),
-            meta: SaveMeta::new(num_of_cache_actor),
-        }
+    pub fn with_vec(meta: SaveMeta) -> Self {
+        Self { target: EncodingTarget::InMemory(Vec::new()), meta }
     }
 
-    pub async fn add_meta(&mut self) -> Result<()> {
-        let meta = [
-            encode_header("0011")?,
-            encode_metadata(Vec::from([("redis-ver", "6.0.16")]))?,
-            encode_database_info(0)?,
-        ];
+    pub async fn encode_meta(&mut self) -> Result<()> {
+        let mut metadata = vec![("redis-ver", "6.0.16")];
+        if &self.meta.repl_id != "?" {
+            metadata.push(("repl-id", &self.meta.repl_id));
+            metadata.push(("repl-offset", &self.meta.offset));
+        }
+
+        let meta = [encode_header("0011")?, encode_metadata(metadata)?, encode_database_info(0)?];
         self.target.write(&meta.concat()).await?;
         Ok(())
     }
@@ -111,16 +110,20 @@ pub struct SaveMeta {
     total_expires_table_size: usize,
     chunk_queue: VecDeque<Vec<CacheEntry>>,
     num_of_cache_actors: usize,
+    repl_id: String,
+    offset: String,
 }
 
 impl SaveMeta {
-    pub fn new(num_of_cache_actors: usize) -> Self {
+    pub fn new(num_of_cache_actors: usize, repl_id: String, offset: String) -> Self {
         Self {
             num_of_saved_table_size_actor: num_of_cache_actors,
             total_key_value_table_size: 0,
             total_expires_table_size: 0,
             chunk_queue: VecDeque::new(),
             num_of_cache_actors,
+            repl_id,
+            offset,
         }
     }
 }
diff --git a/src/services/statefuls/persist/mod.rs b/src/services/statefuls/persist/mod.rs
index b17c6310..808ddbfc 100644
--- a/src/services/statefuls/persist/mod.rs
+++ b/src/services/statefuls/persist/mod.rs
@@ -2,12 +2,11 @@ pub mod actor;
 pub mod encoding_command;
 pub mod endec;
 use super::cache::CacheEntry;
-use std::collections::HashMap;
 
 #[derive(Debug)]
 pub struct DumpFile {
     pub(crate) header: String,
-    pub(crate) metadata: HashMap<String, String>,
+    pub(crate) metadata: DumpMetadata,
     pub(crate) database: Vec<DatabaseSection>,
     pub(crate) checksum: Vec<u8>,
 }
@@ -15,7 +14,7 @@ pub struct DumpFile {
 impl DumpFile {
     pub fn new(
         header: String,
-        metadata: HashMap<String, String>,
+        metadata: DumpMetadata,
         database: Vec<DatabaseSection>,
         checksum: Vec<u8>,
     ) -> Self {
@@ -24,6 +23,20 @@ impl DumpFile {
     pub fn key_values(self) -> Vec<CacheEntry> {
         self.database.into_iter().flat_map(|section| section.storage.into_iter()).collect()
     }
+
+    pub fn extract_replication_info(&self) -> Option<(String, u64)> {
+        match (self.metadata.repl_id.as_ref(), self.metadata.repl_offset) {
+            (Some(repl_id), Some(offset)) => Some((repl_id.clone(), offset)),
+            _ => None,
+        }
+    }
+}
+
+// TODO make it non-nullable?
+#[derive(Debug, Default, PartialEq)]
+pub struct DumpMetadata {
+    pub(crate) repl_id: Option<String>,
+    pub(crate) repl_offset: Option<u64>,
 }
 
 #[derive(Debug)]


Test patch:
diff --git a/tests/test_handshake.rs b/tests/test_handshake.rs
index 0bf19ead..5f4cb894 100644
--- a/tests/test_handshake.rs
+++ b/tests/test_handshake.rs
@@ -9,6 +9,7 @@
 use common::{spawn_server_as_slave, spawn_server_process};
 use duva::client_utils::ClientStreamHandler;
 
+
 mod common;
 
 #[tokio::test]
@@ -27,9 +28,9 @@ async fn test_master_threeway_handshake() {
             format!(
                 "*3\r\n$8\r\nREPLCONF\r\n$14\r\nlistening-port\r\n$4\r\n{client_fake_port}\r\n",
             )
-            .as_bytes(),
+                .as_bytes(),
         )
-        .await,
+            .await,
         "+OK\r\n"
     );
 
@@ -42,12 +43,14 @@ async fn test_master_threeway_handshake() {
     // WHEN - client sends PSYNC command
     // ! The first argument is the replication ID of the master
     // ! Since this is the first time the replica is connecting to the master, the replication ID will be ? (a question mark)
-    h.send(b"*3\r\n$5\r\nPSYNC\r\n$1\r\n?\r\n$2\r\n-1\r\n").await;
-    assert!(h
+    let response = h
         .send_and_get(b"*3\r\n$5\r\nPSYNC\r\n$1\r\n?\r\n$2\r\n-1\r\n")
-        .await
-        // dummy response
-        .starts_with("+FULLRESYNC 8371b4fb1155b71f4a04d3e1bc3e18c4a990aeeb 0\r\n"));
+        .await;
+   
+    let split: Vec<&str> = response.split(' ').collect();
+    assert_eq!(split[0], "+FULLRESYNC");
+    assert_eq!(split[1].len(), 40); // replid length == 40
+    assert_eq!(split[2].len(), 3); // offset(1) + \r\n(2)
 }
 
 #[tokio::test]
diff --git a/tests/test_replication_info.rs b/tests/test_replication_info.rs
index 29a57d97..39744f0a 100644
--- a/tests/test_replication_info.rs
+++ b/tests/test_replication_info.rs
@@ -7,7 +7,7 @@ mod common;
 
 use common::{array, spawn_server_process};
 
-use duva::{client_utils::ClientStreamHandler, services::query_io::QueryIO};
+use duva::client_utils::ClientStreamHandler;
 
 #[tokio::test]
 async fn test_replication_info() {
@@ -19,5 +19,17 @@ async fn test_replication_info() {
     let res = h.send_and_get(&array(vec!["INFO", "replication"])).await;
 
     // THEN
-    assert_eq!(res, QueryIO::BulkString("role:master\r\nconnected_slaves:0\r\nmaster_replid:8371b4fb1155b71f4a04d3e1bc3e18c4a990aeeb\r\nmaster_repl_offset:0\r\nsecond_repl_offset:-1\r\nrepl_backlog_active:0\r\nrepl_backlog_size:1048576\r\nrepl_backlog_first_byte_offset:0\r\nself_identifier:127.0.0.1:49152".into()).serialize());
+    let info: Vec<&str> = res.split("\r\n").collect();
+
+    // THEN
+    assert_eq!(info[0], "$249");
+    assert_eq!(info[1], "role:master");
+    assert_eq!(info[2], "connected_slaves:0");
+    assert!(info[3].starts_with("master_replid:"));
+    assert_eq!(info[4], "master_repl_offset:0");
+    assert_eq!(info[5], "second_repl_offset:-1");
+    assert_eq!(info[6], "repl_backlog_active:0");
+    assert_eq!(info[7], "repl_backlog_size:1048576");
+    assert_eq!(info[8], "repl_backlog_first_byte_offset:0");
+    assert_eq!(info[9], "self_identifier:127.0.0.1:49152");
 }
diff --git a/tests/test_save_read_dump.rs b/tests/test_save_read_dump.rs
index 3d9d8adf..49c80293 100644
--- a/tests/test_save_read_dump.rs
+++ b/tests/test_save_read_dump.rs
@@ -70,6 +70,14 @@ async fn test_save_read_dump() {
     // check keys
     assert_eq!(h.send_and_get(&array(vec!["KEYS", "*"])).await, array(vec!["foo2", "foo"]));
 
+    // check replication info
+    let res = h.send_and_get(&array(vec!["INFO", "replication"])).await;
+    let info: Vec<&str> = res.split("\r\n").collect();
+    let prev_master_repl_id = info[3].split(":").collect::<Vec<&str>>()[1];
+    let prev_master_repl_offset = info[4].split(":").collect::<Vec<&str>>()[1];
+
+    // THEN
+
     // THEN
     assert_eq!(h.send_and_get(&array(vec!["SAVE"])).await, QueryIO::Null.serialize());
 
@@ -77,6 +85,16 @@ async fn test_save_read_dump() {
     tokio::time::sleep(tokio::time::Duration::from_secs(1)).await;
     // keys
     assert_eq!(h.send_and_get(&array(vec!["KEYS", "*"])).await, array(vec!["foo2", "foo"]));
+
+    // replication info
+    let res = h.send_and_get(&array(vec!["INFO", "replication"])).await;
+    let info: Vec<&str> = res.split("\r\n").collect();
+    let master_repl_id = info[3].split(":").collect::<Vec<&str>>()[1];
+    let master_repl_offset = info[4].split(":").collect::<Vec<&str>>()[1];
+
+    // THEN
+    assert_eq!(master_repl_id, prev_master_repl_id);
+    assert_eq!(master_repl_offset, prev_master_repl_offset);
 }
 
 fn create_unique_file_name(function_name: &str) -> String {