From 56486a60c63fa5da1b8fdb877306c725058b1892 Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?=E5=BC=A0=E7=82=8E=E6=B3=BC?= <drdr.xp@gmail.com>
Date: Sat, 29 Oct 2022 00:18:47 +0800
Subject: [PATCH] Fix: Error After change_membership: `assertion failed: value
 > prev`: #584

Problem:

Error After change_membership: `assertion failed: value > prev`,
when changing membership by converting a learner to a voter.

Because the replication streams are re-spawned, thus progress reverts to
zero. Then a reverted progress causes the panic.

Solution:

When re-spawning replications, remember the previous progress.

- Fix: #584
---
 openraft/src/core/raft_core.rs  | 7 ++++---
 openraft/src/replication/mod.rs | 3 ++-
 2 files changed, 6 insertions(+), 4 deletions(-)

diff --git a/openraft/src/core/raft_core.rs b/openraft/src/core/raft_core.rs
index 27ea8ff20..4673c458e 100644
--- a/openraft/src/core/raft_core.rs
+++ b/openraft/src/core/raft_core.rs
@@ -994,6 +994,7 @@ impl<C: RaftTypeConfig, N: RaftNetworkFactory<C>, S: RaftStorage<C>> RaftCore<C,
     pub(crate) async fn spawn_replication_stream(
         &mut self,
         target: C::NodeId,
+        matched: Option<LogId<C::NodeId>>,
     ) -> Result<ReplicationStream<C::NodeId>, N::ConnectionError> {
         let target_node = self.engine.state.membership_state.effective.get_node(&target);
         let membership_log_id = self.engine.state.membership_state.effective.log_id;
@@ -1007,6 +1008,7 @@ impl<C: RaftTypeConfig, N: RaftNetworkFactory<C>, S: RaftStorage<C>> RaftCore<C,
             self.config.clone(),
             self.engine.state.last_log_id(),
             self.engine.state.committed,
+            matched,
             network,
             self.storage.get_log_reader().await,
             self.tx_api.clone(),
@@ -1611,9 +1613,8 @@ impl<C: RaftTypeConfig, N: RaftNetworkFactory<C>, S: RaftStorage<C>> RaftRuntime
             Command::UpdateReplicationStreams { targets } => {
                 self.remove_all_replication().await;
 
-                // TODO: use _matched to initialize replication
-                for (node_id, _matched) in targets.iter() {
-                    match self.spawn_replication_stream(*node_id).await {
+                for (node_id, matched) in targets.iter() {
+                    match self.spawn_replication_stream(*node_id, *matched).await {
                         Ok(state) => {
                             if let Some(l) = &mut self.leader_data {
                                 l.nodes.insert(*node_id, state);
diff --git a/openraft/src/replication/mod.rs b/openraft/src/replication/mod.rs
index db1075ad6..ccb522fd0 100644
--- a/openraft/src/replication/mod.rs
+++ b/openraft/src/replication/mod.rs
@@ -140,6 +140,7 @@ impl<C: RaftTypeConfig, N: RaftNetworkFactory<C>, S: RaftStorage<C>> Replication
         config: Arc<Config>,
         last_log: Option<LogId<C::NodeId>>,
         committed: Option<LogId<C::NodeId>>,
+        matched: Option<LogId<C::NodeId>>,
         network: N::Network,
         log_reader: S::LogReader,
         raft_core_tx: mpsc::UnboundedSender<RaftMsg<C, N, S>>,
@@ -158,7 +159,7 @@ impl<C: RaftTypeConfig, N: RaftNetworkFactory<C>, S: RaftStorage<C>> Replication
             config,
             target_repl_state: TargetReplState::LineRate,
             committed,
-            matched: None,
+            matched,
             max_possible_matched_index: last_log.index(),
             raft_core_tx,
             repl_rx,