connection offset

Migorithm · Migorithm · commit 4e60ff09ae15 · 2025-11-13T20:43:50.000+04:00
diff --git a/duva-client/src/broker/mod.rs b/duva-client/src/broker/mod.rs
@@ -102,7 +102,7 @@ impl Broker {
                     } else {
                         context.callback(ServerResponse::Err {
                             reason: "Failed to route command. Try again after ttl time".to_string(),
-                            request_id: 0,
+                            conn_offset: 0,
                         })
                     };
                 },
@@ -292,7 +292,10 @@ impl Broker {
             // ! otherwise, server will not be able to process the next command
 
             match res {
-                ServerResponse::ReadRes { res: QueryIO::BulkString(..), request_id } => {
+                ServerResponse::ReadRes {
+                    res: QueryIO::BulkString(..),
+                    conn_offset: request_id,
+                } => {
                     connection.request_id = connection.request_id.max(*request_id);
                 },
                 ServerResponse::WriteRes { res: QueryIO::BulkString(..), log_index, .. } => {
diff --git a/duva-client/src/broker/node_connections.rs b/duva-client/src/broker/node_connections.rs
@@ -97,7 +97,8 @@ impl NodeConnection {
     }
 
     pub(crate) async fn send(&self, client_action: ClientAction) -> anyhow::Result<()> {
-        let session_request = SessionRequest { request_id: self.request_id, action: client_action };
+        let session_request =
+            SessionRequest { conn_offset: self.request_id, action: client_action };
         self.writer
             .send(MsgToServer::Command(bincode::encode_to_vec(session_request, SERDE_CONFIG)?))
             .await
diff --git a/duva-client/src/command.rs b/duva-client/src/command.rs
@@ -35,7 +35,7 @@ impl CommandQueue {
 
         let result = context
             .get_result()
-            .unwrap_or_else(|err| ServerResponse::Err { reason: err.to_string(), request_id: 0 });
+            .unwrap_or_else(|err| ServerResponse::Err { reason: err.to_string(), conn_offset: 0 });
         context.callback(result);
     }
 }
@@ -84,19 +84,21 @@ impl InputContext {
             ClientAction::NonMutating(Keys { pattern: _ } | MGet { keys: _ }) => {
                 let mut init = QueryIO::Array(Vec::with_capacity(iterator.len()));
 
-                while let Some(ServerResponse::ReadRes { res, request_id }) = iterator.next() {
+                while let Some(ServerResponse::ReadRes { res, conn_offset: request_id }) =
+                    iterator.next()
+                {
                     init = init.merge(res)?;
                     highest_req_id = highest_req_id.max(request_id);
                 }
-                Ok(ServerResponse::ReadRes { res: init, request_id: highest_req_id })
+                Ok(ServerResponse::ReadRes { res: init, conn_offset: highest_req_id })
             },
 
             ClientAction::NonMutating(Exists { keys: _ }) => {
                 let mut count = 0;
 
                 while let Some(ServerResponse::ReadRes {
                     res: QueryIO::BulkString(byte),
-                    request_id,
+                    conn_offset: request_id,
                 }) = iterator.next()
                 {
                     let num = String::from_utf8(byte.to_vec())
@@ -109,15 +111,15 @@ impl InputContext {
 
                 Ok(ServerResponse::ReadRes {
                     res: QueryIO::BulkString(BinBytes::new(count.to_string())),
-                    request_id: highest_req_id,
+                    conn_offset: highest_req_id,
                 })
             },
             ClientAction::Mutating(LogEntry::Delete { keys: _ }) => {
                 let mut count = 0;
 
                 while let Some(ServerResponse::WriteRes {
                     res: QueryIO::BulkString(value),
-                    request_id,
+                    conn_offset: request_id,
                     ..
                 }) = iterator.next()
                 {
@@ -128,7 +130,7 @@ impl InputContext {
                 Ok(ServerResponse::WriteRes {
                     res: QueryIO::BulkString(BinBytes::new(count.to_string())),
                     log_index: 0, // TODO
-                    request_id: highest_req_id,
+                    conn_offset: highest_req_id,
                 })
             },
             _ => iterator.next().ok_or(anyhow::anyhow!("Expected exactly one result")),
diff --git a/duva/src/domains/cluster_actors/actor.rs b/duva/src/domains/cluster_actors/actor.rs
@@ -1,6 +1,6 @@
 use super::ClusterCommand;
-use super::ConsensusClientResponse;
-use super::ConsensusReq;
+use super::ConsensusRequest;
+use super::ConsensusResponse;
 use super::LazyOption;
 use super::hash_ring::HashRing;
 pub mod client_sessions;
@@ -325,9 +325,9 @@ impl<T: TWriteAheadLog> ClusterActor<T> {
         }
     }
 
-    pub(crate) async fn leader_req_consensus(&mut self, req: ConsensusReq) {
+    pub(crate) async fn leader_req_consensus(&mut self, req: ConsensusRequest) {
         if !self.replication.is_leader() {
-            req.callback.send(ConsensusClientResponse::Result {
+            req.callback.send(ConsensusResponse::Result {
                 res: Err(anyhow::anyhow!("Write given to follower")),
                 log_index: self.replication.last_log_index(),
             });
@@ -337,7 +337,7 @@ impl<T: TWriteAheadLog> ClusterActor<T> {
         if self.client_sessions.is_processed(&req.conn_offset) {
             // mapping between early returned values to client result
             let key = req.entry.all_keys().into_iter().map(String::from).collect();
-            req.callback.send(ConsensusClientResponse::AlreadyProcessed {
+            req.callback.send(ConsensusResponse::AlreadyProcessed {
                 key,
                 // TODO : remove unwrap
                 request_id: req.conn_offset.unwrap().offset,
@@ -358,22 +358,22 @@ impl<T: TWriteAheadLog> ClusterActor<T> {
                 // To notify client's of what keys have been moved.
                 // ! Still, client won't know where the key has been moved. The assumption here is client SHOULD have correct hashring information.
                 let moved_keys = replids.except(&self.log_state().replid).join(" ");
-                req.callback.send(ConsensusClientResponse::Result {
+                req.callback.send(ConsensusResponse::Result {
                     res: Err(anyhow::anyhow!("Moved! {moved_keys}")),
                     log_index: self.replication.last_log_index(),
                 })
             },
             Err(err) => {
                 err!("{}", err);
-                req.callback.send(ConsensusClientResponse::Result {
+                req.callback.send(ConsensusResponse::Result {
                     res: Err(anyhow::anyhow!(err)),
                     log_index: self.replication.last_log_index(),
                 });
             },
         }
     }
 
-    async fn req_consensus(&mut self, req: ConsensusReq, send_in_mills: Option<u64>) {
+    async fn req_consensus(&mut self, req: ConsensusRequest, send_in_mills: Option<u64>) {
         let log_index = self.replication.write_single_entry(
             req.entry,
             self.log_state().term,
@@ -387,7 +387,7 @@ impl<T: TWriteAheadLog> ClusterActor<T> {
             self.replication.increase_con_idx_by(1);
             let _ = self.replication.flush();
             let res = self.commit_entry(entry.entry, log_index).await;
-            req.callback.send(ConsensusClientResponse::Result { res, log_index });
+            req.callback.send(ConsensusResponse::Result { res, log_index });
             return;
         }
         self.consensus_tracker
@@ -959,7 +959,7 @@ impl<T: TWriteAheadLog> ClusterActor<T> {
         let res = self.commit_entry(log_entry.entry, log_index).await;
         let _ = self.replication.flush();
 
-        voting.callback.send(ConsensusClientResponse::Result { res, log_index });
+        voting.callback.send(ConsensusResponse::Result { res, log_index });
     }
 
     async fn commit_entry(&mut self, entry: LogEntry, index: u64) -> anyhow::Result<QueryIO> {
@@ -1340,7 +1340,7 @@ impl<T: TWriteAheadLog> ClusterActor<T> {
         }
 
         let (callback, rx) = Callback::create();
-        let req = ConsensusReq {
+        let req = ConsensusRequest {
             entry: LogEntry::MSet { entries: migrate_batch.entries.clone() },
             callback,
             conn_offset: None,
@@ -1365,7 +1365,7 @@ impl<T: TWriteAheadLog> ClusterActor<T> {
         });
     }
 
-    async fn make_consensus_in_batch(&mut self, req: ConsensusReq) {
+    async fn make_consensus_in_batch(&mut self, req: ConsensusRequest) {
         self.req_consensus(req, None).await;
         let _ = self.replication.flush();
         self.send_rpc().await;
@@ -1384,7 +1384,7 @@ impl<T: TWriteAheadLog> ClusterActor<T> {
 
         // make consensus request for delete
         let (callback, rx) = Callback::create();
-        let req = ConsensusReq {
+        let req = ConsensusRequest {
             entry: LogEntry::Delete { keys: pending_migration_batch.keys.clone() },
             callback,
             conn_offset: None,
diff --git a/duva/src/domains/cluster_actors/actor/tests/elections.rs b/duva/src/domains/cluster_actors/actor/tests/elections.rs
@@ -233,7 +233,7 @@ async fn test_become_candidate_not_allow_write_request_processing() {
     let (tx, rx) = Callback::create();
     let session_req = ConnectionOffset::new(1, "client1".to_string());
 
-    let consensus_request = ConsensusReq {
+    let consensus_request = ConsensusRequest {
         entry: LogEntry::Set { entry: CacheEntry::new("key".to_string(), "value") },
         callback: tx,
         conn_offset: Some(session_req),
@@ -248,5 +248,5 @@ async fn test_become_candidate_not_allow_write_request_processing() {
     assert!(value.is_null());
 
     let res = rx.0.await.unwrap();
-    assert!(matches!(res, ConsensusClientResponse::Result { res: Err(_), log_index: _ }))
+    assert!(matches!(res, ConsensusResponse::Result { res: Err(_), log_index: _ }))
 }
diff --git a/duva/src/domains/cluster_actors/actor/tests/mod.rs b/duva/src/domains/cluster_actors/actor/tests/mod.rs
@@ -237,10 +237,10 @@ impl Helper {
     }
 
     fn consensus_request(
-        callback: Callback<ConsensusClientResponse>,
+        callback: Callback<ConsensusResponse>,
         session_req: Option<ConnectionOffset>,
-    ) -> ConsensusReq {
-        ConsensusReq {
+    ) -> ConsensusRequest {
+        ConsensusRequest {
             entry: LogEntry::Set { entry: CacheEntry::new("foo".to_string(), "bar") },
             callback,
             conn_offset: session_req,
diff --git a/duva/src/domains/cluster_actors/actor/tests/replications.rs b/duva/src/domains/cluster_actors/actor/tests/replications.rs
@@ -433,7 +433,7 @@ async fn req_consensus_inserts_consensus_voting() {
     let session_request = ConnectionOffset::new(1, client_id);
     let w_req = LogEntry::Set { entry: CacheEntry::new("foo".to_string(), "bar") };
     let consensus_request =
-        ConsensusReq { entry: w_req.clone(), callback, conn_offset: Some(session_request.clone()) };
+        ConsensusRequest { entry: w_req.clone(), callback, conn_offset: Some(session_request.clone()) };
 
     // WHEN
     leader_c_actor.req_consensus(consensus_request, None).await;
@@ -462,7 +462,7 @@ async fn test_leader_req_consensus_early_return_when_already_processed_session_r
     tokio::spawn(cluster_actor.handle());
     let (callback, rx) = Callback::create();
     handler
-        .send(ClusterCommand::Client(ClusterClientRequest::MakeConsensus(ConsensusReq {
+        .send(ClusterCommand::Client(ClusterClientRequest::MakeConsensus(ConsensusRequest {
             entry: LogEntry::Set { entry: CacheEntry::new("foo".to_string(), "bar") },
             callback,
             conn_offset: Some(client_req),
@@ -619,7 +619,7 @@ async fn test_leader_req_consensus_with_processed_session() {
 
     // WHEN - send request with already processed session
     let (tx, rx) = Callback::create();
-    let consensus_request = ConsensusReq {
+    let consensus_request = ConsensusRequest {
         entry: LogEntry::Set { entry: CacheEntry::new("test_key".to_string(), "test_value") },
         callback: tx,
         conn_offset: Some(session_req),
@@ -632,7 +632,7 @@ async fn test_leader_req_consensus_with_processed_session() {
     assert_eq!(cluster_actor.log_state().last_log_index, 0);
 
     // Verify the response indicates already processed
-    let ConsensusClientResponse::AlreadyProcessed { key, request_id: 1 } = rx.recv().await else {
+    let ConsensusResponse::AlreadyProcessed { key, request_id: 1 } = rx.recv().await else {
         panic!("Expected AlreadyProcessed response");
     };
     assert_eq!(key, vec!["test_key".to_string()]);
diff --git a/duva/src/domains/cluster_actors/command.rs b/duva/src/domains/cluster_actors/command.rs
@@ -57,7 +57,7 @@ pub enum ClusterClientRequest {
     ReplicationState(Callback<ReplicationState>),
     Forget(PeerIdentifier, Callback<Option<()>>),
     ReplicaOf(PeerIdentifier, Callback<anyhow::Result<()>>),
-    MakeConsensus(ConsensusReq),
+    MakeConsensus(ConsensusRequest),
     ClusterNodes(Callback<Vec<ReplicationState>>),
     GetRoles(Callback<Vec<(PeerIdentifier, ReplicationRole)>>),
     SubscribeToTopologyChange(Callback<tokio::sync::broadcast::Receiver<Topology>>),
@@ -74,13 +74,6 @@ impl From<ClusterClientRequest> for ClusterCommand {
     }
 }
 
-#[derive(Debug, PartialEq, Eq)]
-pub(crate) struct ConsensusReq {
-    pub(crate) entry: LogEntry,
-    pub(crate) callback: Callback<ConsensusClientResponse>,
-    pub(crate) conn_offset: Option<ConnectionOffset>,
-}
-
 #[derive(Debug, Clone, PartialEq, Eq, bincode::Encode, bincode::Decode)]
 pub struct ConnectionOffset {
     pub(crate) offset: u64,
@@ -92,8 +85,15 @@ impl ConnectionOffset {
     }
 }
 
+#[derive(Debug, PartialEq, Eq)]
+pub(crate) struct ConsensusRequest {
+    pub(crate) entry: LogEntry,
+    pub(crate) callback: Callback<ConsensusResponse>,
+    pub(crate) conn_offset: Option<ConnectionOffset>,
+}
+
 #[derive(Debug)]
-pub(crate) enum ConsensusClientResponse {
+pub(crate) enum ConsensusResponse {
     AlreadyProcessed { key: Vec<String>, request_id: u64 },
     Result { res: anyhow::Result<QueryIO>, log_index: u64 },
 }
diff --git a/duva/src/domains/peers/command.rs b/duva/src/domains/peers/command.rs
@@ -1,7 +1,7 @@
 use crate::{
     domains::{
         caches::cache_objects::CacheEntry,
-        cluster_actors::{ClusterCommand, ConsensusReq, hash_ring::HashRing},
+        cluster_actors::{ClusterCommand, ConsensusRequest, hash_ring::HashRing},
         replications::{ReplicationId, ReplicationState, WriteOperation},
     },
     prelude::PeerIdentifier,
@@ -188,12 +188,12 @@ pub(crate) struct QueuedKeysToMigrate {
 
 #[derive(Debug, Default)]
 pub(crate) struct PendingRequests {
-    requests: VecDeque<ConsensusReq>,
+    requests: VecDeque<ConsensusRequest>,
     batches: HashMap<BatchId, QueuedKeysToMigrate>,
     pub(crate) callbacks: Vec<Callback<()>>,
 }
 impl PendingRequests {
-    pub(crate) fn add_req(&mut self, req: ConsensusReq) {
+    pub(crate) fn add_req(&mut self, req: ConsensusRequest) {
         self.requests.push_back(req);
     }
     pub(crate) fn store_batch(&mut self, id: BatchId, batch: QueuedKeysToMigrate) {
@@ -202,7 +202,7 @@ impl PendingRequests {
     pub(crate) fn pop_batch(&mut self, id: &BatchId) -> Option<QueuedKeysToMigrate> {
         self.batches.remove(id)
     }
-    pub(crate) fn extract_requests(&mut self) -> VecDeque<ConsensusReq> {
+    pub(crate) fn extract_requests(&mut self) -> VecDeque<ConsensusRequest> {
         std::mem::take(&mut self.requests)
     }
 
diff --git a/duva/src/domains/replications/consensus/log.rs b/duva/src/domains/replications/consensus/log.rs
@@ -1,6 +1,6 @@
 use crate::{
     domains::{
-        cluster_actors::{ConnectionOffset, ConsensusClientResponse},
+        cluster_actors::{ConnectionOffset, ConsensusResponse},
         peers::identifier::PeerIdentifier,
     },
     make_smart_pointer,
@@ -16,13 +16,13 @@ make_smart_pointer!(LogConsensusTracker, HashMap<u64, LogConsensusVoting>);
 #[derive(Debug)]
 pub struct LogConsensusVoting {
     pub(crate) voters: Vec<PeerIdentifier>,
-    pub(crate) callback: Callback<ConsensusClientResponse>,
+    pub(crate) callback: Callback<ConsensusResponse>,
     pub(crate) cnt: u8,
     pub(crate) conn_offset: Option<ConnectionOffset>,
 }
 impl LogConsensusVoting {
     pub(crate) fn new(
-        callback: Callback<ConsensusClientResponse>,
+        callback: Callback<ConsensusResponse>,
         replica_count: usize,
         conn_offset: Option<ConnectionOffset>,
     ) -> Self {
diff --git a/duva/src/lib.rs b/duva/src/lib.rs
@@ -255,7 +255,7 @@ impl StartUpFacade {
                     cache_manager: self.cache_manager.clone(),
                 };
                 tokio::spawn(
-                    ClientStreamReader { client_id, r: read_half }
+                    ClientStreamReader { conn_id: client_id, r: read_half }
                         .handle_client_stream(client_controller, writer.run(observer)),
                 );
             },
diff --git a/duva/src/presentation/clients/controller.rs b/duva/src/presentation/clients/controller.rs
diff --git a/duva/src/presentation/clients/request.rs b/duva/src/presentation/clients/request.rs
diff --git a/duva/src/presentation/clients/stream.rs b/duva/src/presentation/clients/stream.rs

Original file line number	Diff line number	Diff line change
`@@ -97,7 +97,8 @@ impl NodeConnection {`
`97`	`97`	`}`
`98`	`98`
`99`	`99`	`pub(crate) async fn send(&self, client_action: ClientAction) -> anyhow::Result<()> {`
`100`		`- let session_request = SessionRequest { request_id: self.request_id, action: client_action };`
	`100`	`+ let session_request =`
	`101`	`+ SessionRequest { conn_offset: self.request_id, action: client_action };`
`101`	`102`	`self.writer`
`102`	`103`	`.send(MsgToServer::Command(bincode::encode_to_vec(session_request, SERDE_CONFIG)?))`
`103`	`104`	`.await`