refactor(op): add OpPersistPeer and impl OpResult for it

src/main.rs

@@ -307,7 +307,7 @@ fn picolib_setup(args: &args::Run) {
                         entry.term.to_string(),
                         entry
                             .lc()
-                            .map(ToString::to_string)
+                            .map(|lc| lc.to_string())
                             .unwrap_or_else(String::new),
                         entry.payload().to_string(),
                     ];
@@ -829,7 +829,7 @@ fn start_boot(args: &args::Run) {
             .expect("cannot fail")
             .into(),
         );
-        init_entries_push_op(traft::Op::persist_peer(peer));
+        init_entries_push_op(traft::OpPersistPeer::new(peer).into());
         init_entries_push_op(
             OpDML::insert(
                 ClusterwideSpace::State,

src/traft/mod.rs

@@ -102,16 +102,13 @@ pub enum Op {
     /// No operation.
     Nop,
     /// Print the message in tarantool log.
-    Info {
-        msg: String,
-    },
+    Info { msg: String },
     /// Evaluate the code on every instance in cluster.
     EvalLua(OpEvalLua),
     ///
     ReturnOne(OpReturnOne),
-    PersistPeer {
-        peer: Box<Peer>,
-    },
+    /// Update the given peer's entry in [`storage::Peers`].
+    PersistPeer(OpPersistPeer),
     /// Cluster-wide data modification operation.
     /// Should be used to manipulate the cluster-wide configuration.
     Dml(OpDML),
@@ -124,7 +121,7 @@ impl std::fmt::Display for Op {
             Self::Info { msg } => write!(f, "Info({msg:?})"),
             Self::EvalLua(OpEvalLua { code }) => write!(f, "EvalLua({code:?})"),
             Self::ReturnOne(_) => write!(f, "ReturnOne"),
-            Self::PersistPeer { peer } => {
+            Self::PersistPeer(OpPersistPeer(peer)) => {
                 write!(f, "PersistPeer{}", peer)
             }
             Self::Dml(OpDML::Insert { space, tuple }) => {
@@ -174,7 +171,7 @@ impl std::fmt::Display for Op {
 }
 
 impl Op {
-    pub fn on_commit(&self, peers: &storage::Peers) -> Box<dyn AnyWithTypeName> {
+    pub fn on_commit(self, peers: &storage::Peers) -> Box<dyn AnyWithTypeName> {
         match self {
             Self::Nop => Box::new(()),
             Self::Info { msg } => {
@@ -183,30 +180,24 @@ impl Op {
             }
             Self::EvalLua(op) => Box::new(op.result()),
             Self::ReturnOne(op) => Box::new(op.result()),
-            Self::PersistPeer { peer } => {
-                peers.put(peer).unwrap();
-                Box::new(peer.clone())
+            Self::PersistPeer(op) => {
+                let peer = op.result();
+                peers.put(&peer).unwrap();
+                peer
             }
             Self::Dml(op) => {
-                let res = Box::new(op.result());
                 if op.space() == &ClusterwideSpace::State {
                     event::broadcast(Event::ClusterStateChanged);
                 }
-                res
+                Box::new(op.result())
             }
         }
     }
-
-    pub fn persist_peer(peer: Peer) -> Self {
-        Self::PersistPeer {
-            peer: Box::new(peer),
-        }
-    }
 }
 
 impl OpResult for Op {
     type Result = ();
-    fn result(&self) -> Self::Result {}
+    fn result(self) -> Self::Result {}
 }
 
 impl From<OpReturnOne> for Op {
@@ -220,7 +211,7 @@ pub struct OpReturnOne;
 
 impl OpResult for OpReturnOne {
     type Result = u8;
-    fn result(&self) -> Self::Result {
+    fn result(self) -> Self::Result {
         1
     }
 }
@@ -232,7 +223,7 @@ pub struct OpEvalLua {
 
 impl OpResult for OpEvalLua {
     type Result = StdResult<(), LuaError>;
-    fn result(&self) -> Self::Result {
+    fn result(self) -> Self::Result {
         crate::tarantool::exec(&self.code)
     }
 }
@@ -245,7 +236,37 @@ impl From<OpEvalLua> for Op {
 
 pub trait OpResult {
     type Result: 'static;
-    fn result(&self) -> Self::Result;
+    // FIXME: this signature makes it look like result of any operation depends
+    // only on what is contained within the operation which is almost never true
+    // And it makes it hard to do anything useful inside this function.
+    fn result(self) -> Self::Result;
+}
+
+////////////////////////////////////////////////////////////////////////////////
+// OpPersistPeer
+////////////////////////////////////////////////////////////////////////////////
+
+#[derive(Clone, Debug, Serialize, Deserialize, PartialEq, Eq)]
+pub struct OpPersistPeer(pub Box<Peer>);
+
+impl OpPersistPeer {
+    pub fn new(peer: Peer) -> Self {
+        Self(Box::new(peer))
+    }
+}
+
+impl OpResult for OpPersistPeer {
+    type Result = Box<Peer>;
+    fn result(self) -> Self::Result {
+        self.0
+    }
+}
+
+impl From<OpPersistPeer> for Op {
+    #[inline]
+    fn from(op: OpPersistPeer) -> Op {
+        Op::PersistPeer(op)
+    }
 }
 
 //////////////////////////////////////////////////////////////////////////////////////////
@@ -280,12 +301,12 @@ pub enum OpDML {
 
 impl OpResult for OpDML {
     type Result = tarantool::Result<Option<Tuple>>;
-    fn result(&self) -> Self::Result {
+    fn result(self) -> Self::Result {
         match self {
-            Self::Insert { space, tuple } => space.insert(tuple).map(Some),
-            Self::Replace { space, tuple } => space.replace(tuple).map(Some),
-            Self::Update { space, key, ops } => space.update(key, ops),
-            Self::Delete { space, key } => space.delete(key),
+            Self::Insert { space, tuple } => space.insert(&tuple).map(Some),
+            Self::Replace { space, tuple } => space.replace(&tuple).map(Some),
+            Self::Update { space, key, ops } => space.update(&key, &ops),
+            Self::Delete { space, key } => space.delete(&key),
         }
     }
 }
@@ -619,18 +640,19 @@ impl ContextCoercion for EntryContextConfChange {}
 
 impl Entry {
     /// Returns the logical clock value if it's an `EntryNormal`.
-    pub fn lc(&self) -> Option<&LogicalClock> {
+    pub fn lc(&self) -> Option<LogicalClock> {
         match &self.context {
-            Some(EntryContext::Normal(v)) => Some(&v.lc),
+            Some(EntryContext::Normal(v)) => Some(v.lc),
             Some(EntryContext::ConfChange(_)) => None,
             None => None,
         }
     }
 
-    /// Returns the contained `Op` if it's an `EntryNormal`.
-    fn op(&self) -> Option<&Op> {
-        match &self.context {
-            Some(EntryContext::Normal(v)) => Some(&v.op),
+    /// Returns the contained `Op` if it's an `EntryNormal`
+    /// consuming `self` by value.
+    fn into_op(self) -> Option<Op> {
+        match self.context {
+            Some(EntryContext::Normal(v)) => Some(v.op),
             Some(EntryContext::ConfChange(_)) => None,
             None => None,
         }

src/traft/node.rs

@@ -37,11 +37,11 @@ use crate::storage::{Clusterwide, ClusterwideSpace, StateKey};
 use crate::stringify_cfunc;
 use crate::traft::rpc;
 use crate::traft::ContextCoercion as _;
-use crate::traft::OpDML;
 use crate::traft::Peer;
 use crate::traft::RaftId;
 use crate::traft::RaftIndex;
 use crate::traft::RaftTerm;
+use crate::traft::{OpDML, OpPersistPeer};
 use crate::unwrap_some_or;
 use crate::warn_or_panic;
 use ::tarantool::util::IntoClones as _;
@@ -278,7 +278,7 @@ impl Node {
                 // FIXME: this error should be handled
                 let _ = notify.recv_any().await;
             }
-            notify.recv().await
+            notify.recv().await.map(Box::new)
         })
     }
 
@@ -438,7 +438,7 @@ impl NodeImpl {
         T: Into<traft::Op>,
     {
         let (lc, notify) = self.schedule_notification();
-        let ctx = traft::EntryContextNormal::new(lc, op);
+        let ctx = traft::EntryContextNormal::new(lc, op.into());
         self.raw_node.propose(ctx.to_bytes(), vec![])?;
         Ok(notify)
     }
@@ -513,7 +513,7 @@ impl NodeImpl {
             }
         };
         let (lc, notify) = self.schedule_notification();
-        let ctx = traft::EntryContextNormal::new(lc, Op::persist_peer(peer));
+        let ctx = traft::EntryContextNormal::new(lc, OpPersistPeer::new(peer));
 
         // Important! Calling `raw_node.propose()` may result in
         // `ProposalDropped` error, but the topology has already been
@@ -635,18 +635,11 @@ impl NodeImpl {
         expelled: &mut bool,
     ) {
         assert_eq!(entry.entry_type, raft::EntryType::EntryNormal);
-        let result = entry
-            .op()
-            .unwrap_or(&traft::Op::Nop)
-            .on_commit(&self.storage.peers);
+        let lc = entry.lc();
+        let index = entry.index;
+        let op = entry.into_op().unwrap_or(traft::Op::Nop);
 
-        if let Some(lc) = entry.lc() {
-            if let Some(notify) = self.notifications.remove(lc) {
-                notify.notify_ok_any(result);
-            }
-        }
-
-        if let Some(traft::Op::PersistPeer { peer, .. }) = entry.op() {
+        if let traft::Op::PersistPeer(OpPersistPeer(peer)) = &op {
             *topology_changed = true;
             if peer.current_grade == CurrentGradeVariant::Expelled && peer.raft_id == self.raft_id()
             {
@@ -655,7 +648,16 @@ impl NodeImpl {
             }
         }
 
-        if let Some(notify) = self.joint_state_latch.take_or_keep(&entry.index) {
+        // apply the operation
+        let result = op.on_commit(&self.storage.peers);
+
+        if let Some(lc) = &lc {
+            if let Some(notify) = self.notifications.remove(lc) {
+                notify.notify_ok_any(result);
+            }
+        }
+
+        if let Some(notify) = self.joint_state_latch.take_or_keep(&index) {
             // It was expected to be a ConfChange entry, but it's
             // normal. Raft must have overriden it, or there was
             // a re-election.