feat: add ChangeMembers::Batch(Vec<ChangeMembers>) for batched membership changes

This commit introduces the `ChangeMembers::Batch(Vec<ChangeMembers>)`
variant, allowing `Raft::change_membership()` to accept an arbitrary
series of membership configuration changes as a single input command.
This enables users to define complex membership changes by combining
multiple `ChangeMembers` operations.

### Example:
The following example demonstrates removing nodes 1 and 2, and adding
node 3 in a two-step joint membership configuration change:

```rust
my_raft.change_membership(ChangeMembers::Batch(vec![
    ChangeMembers::RemoveVoters(btreeset! {1, 2}),
    ChangeMembers::AddVoters(btreemap! {3 => ()}),
]));
```

Author: drmingdrmer

openraft/src/change_members.rs

@@ -55,6 +55,12 @@ where C: RaftTypeConfig
     /// Every voter has to have a corresponding node in the new
     /// set, otherwise it returns [`error::LearnerNotFound`](`crate::error::LearnerNotFound`) error.
     ReplaceAllNodes(BTreeMap<C::NodeId, C::Node>),
+
+    /// Apply multiple changes to membership config.
+    ///
+    /// The changes are applied in the order they are given.
+    /// And it still finishes in a two-step joint config change.
+    Batch(Vec<ChangeMembers<C>>),
 }
 
 /// Convert a series of ids to a `Replace` operation.

openraft/src/core/sm/worker.rs

@@ -20,6 +20,8 @@ use crate::entry::RaftEntry;
 use crate::entry::RaftPayload;
 use crate::raft::responder::Responder;
 use crate::raft::ClientWriteResponse;
+#[cfg(doc)]
+use crate::storage::RaftLogStorage;
 use crate::storage::RaftStateMachine;
 use crate::storage::Snapshot;
 use crate::type_config::alias::JoinHandleOf;
@@ -43,8 +45,6 @@ where
     state_machine: SM,
 
     /// Read logs from the [`RaftLogStorage`] implementation to apply them to state machine.
-    ///
-    /// [`RaftLogStorage`]: `crate::storage::RaftLogStorage`
     log_reader: LR,
 
     /// Read command from RaftCore to execute.

openraft/src/membership/membership.rs

@@ -315,26 +315,59 @@ where C: RaftTypeConfig
     pub(crate) fn change(mut self, change: ChangeMembers<C>, retain: bool) -> Result<Self, ChangeMembershipError<C>> {
         tracing::debug!(change = debug(&change), "{}", func_name!());
 
+        let Membership { mut configs, nodes } = self.clone().compute_target_membership(change);
+
+        // Safe unwrap(): `calculate_goal()` yields a uniform config.
+        let target_voter_ids = configs.pop().unwrap();
+
+        self.nodes = nodes;
+        let new_membership = self.next_coherent(target_voter_ids, retain);
+
+        tracing::debug!(new_membership = display(&new_membership), "new membership");
+
+        new_membership.ensure_valid()?;
+
+        Ok(new_membership)
+    }
+
+    /// Compute the target membership configuration by applying a membership change.
+    ///
+    /// This method:
+    /// - Uses only the last config entry from the current membership. If there are multiple
+    ///   entries, it indicates an ongoing joint consensus change. The last entry represents the
+    ///   target configuration toward which the cluster is transitioning.
+    /// - Applies the specified membership change to create a new target configuration
+    /// - Returns a new `Membership` with the target voter IDs and nodes
+    ///
+    /// Note: This is an intermediate step in membership changes. The result may need to be
+    /// transformed into a coherent configuration before being applied.
+    fn compute_target_membership(mut self, change: ChangeMembers<C>) -> Membership<C> {
         let last = self.get_joint_config().last().cloned().unwrap_or_default();
 
-        let new_membership = match change {
+        match change {
             ChangeMembers::AddVoterIds(add_voter_ids) => {
                 let new_voter_ids = last.union(&add_voter_ids).cloned().collect::<BTreeSet<_>>();
-                self.next_coherent(new_voter_ids, retain)
+                self.configs = vec![new_voter_ids];
+                self
             }
             ChangeMembers::AddVoters(add_voters) => {
                 // Add nodes without overriding existent
                 self.nodes = Self::extend_nodes(self.nodes, &add_voters);
 
                 let add_voter_ids = add_voters.keys().cloned().collect::<BTreeSet<_>>();
                 let new_voter_ids = last.union(&add_voter_ids).cloned().collect::<BTreeSet<_>>();
-                self.next_coherent(new_voter_ids, retain)
+                self.configs = vec![new_voter_ids];
+                self
             }
             ChangeMembers::RemoveVoters(remove_voter_ids) => {
                 let new_voter_ids = last.difference(&remove_voter_ids).cloned().collect::<BTreeSet<_>>();
-                self.next_coherent(new_voter_ids, retain)
+                self.configs = vec![new_voter_ids];
+                self
+            }
+            ChangeMembers::ReplaceAllVoters(all_voter_ids) => {
+                self.configs = vec![all_voter_ids];
+                self
             }
-            ChangeMembers::ReplaceAllVoters(all_voter_ids) => self.next_coherent(all_voter_ids, retain),
             ChangeMembers::AddNodes(add_nodes) => {
                 // When adding nodes, do not override existing node
                 for (node_id, node) in add_nodes.into_iter() {
@@ -358,13 +391,13 @@ where C: RaftTypeConfig
                 self.nodes = all_nodes;
                 self
             }
-        };
-
-        tracing::debug!(new_membership = display(&new_membership), "new membership");
-
-        new_membership.ensure_valid()?;
-
-        Ok(new_membership)
+            ChangeMembers::Batch(batch) => {
+                for change in batch {
+                    self = self.compute_target_membership(change);
+                }
+                self
+            }
+        }
     }
 
     /// Build a QuorumSet from current joint config
@@ -597,4 +630,39 @@ mod tests {
 
         Ok(())
     }
+
+    /// Test membership change desribed by a batch operations.
+    ///
+    /// The batch operations add one voter and remove another.
+    /// It still finish in a two step joint config change.
+    #[test]
+    fn test_membership_change_batch() -> anyhow::Result<()> {
+        let m = || Membership::<UTConfig> {
+            configs: vec![btreeset! {1,2}],
+            nodes: btreemap! {1=>(),2=>(),3=>()},
+        };
+
+        let rm_2_add_5 = || {
+            ChangeMembers::Batch(vec![
+                ChangeMembers::RemoveVoters(btreeset! {2}),
+                ChangeMembers::AddVoters(btreemap! {5=>()}),
+            ])
+        };
+
+        let step1 = m().change(rm_2_add_5(), false)?;
+
+        assert_eq!(step1, Membership::<UTConfig> {
+            configs: vec![btreeset! {1,2}, btreeset! {1,5}],
+            nodes: btreemap! {1=>(),2=>(),3=>(),5=>()}
+        });
+
+        let step2 = step1.change(rm_2_add_5(), false)?;
+
+        assert_eq!(step2, Membership::<UTConfig> {
+            configs: vec![btreeset! {1,5}],
+            nodes: btreemap! {1=>(),3=>(), 5=>()}
+        });
+
+        Ok(())
+    }
 }

openraft/src/membership/membership_test.rs

@@ -141,14 +141,17 @@ fn test_membership_add_learner() -> anyhow::Result<()> {
     // Add learner that presents in old cluster has no effect.
 
     let res = m_1_2.clone().change(ChangeMembers::AddNodes(btreemap! {1=>node("3")}), true)?;
-    assert_eq!(m_1_2, res);
+    assert_eq!(
+        Membership::<UTConfig<TestNode>>::new_unchecked(vec![btreeset! {2}], btreemap! {1=>node("1"), 2=>node("2")},),
+        res
+    );
 
     // Success to add a learner
 
     let m_1_2_3 = m_1_2.change(ChangeMembers::AddNodes(btreemap! {3=>node("3")}), true)?;
     assert_eq!(
         Membership::<UTConfig<TestNode>>::new_unchecked(
-            vec![btreeset! {1}, btreeset! {2}],
+            vec![btreeset! {2}],
             btreemap! {1=>node("1"), 2=>node("2"), 3=>node("3")}
         ),
         m_1_2_3
@@ -172,7 +175,7 @@ fn test_membership_update_nodes() -> anyhow::Result<()> {
     let m_1_2_3 = m_1_2.change(ChangeMembers::SetNodes(btreemap! {2=>node("20"), 3=>node("30")}), true)?;
     assert_eq!(
         Membership::<UTConfig<TestNode>>::new_unchecked(
-            vec![btreeset! {1}, btreeset! {2}],
+            vec![btreeset! {2}],
             btreemap! {1=>node("1"), 2=>node("20"), 3=>node("30")}
         ),
         m_1_2_3