Change: RaftStorage: use EffectiveMembership instead of Option<_>

The value of membership config of an uninitialized raft-node should be:
```rust
EffectiveMembership {
    log_id: None,
    membership: Membership::default(),
}

```

I.e., there is always a membership config for either an initialized or
an uninitialized raft-node.
Because `EffectiveMembership` is not a value that could be `None`, but it is more like a container of values.

Changed struct:

```rust
struct InitialState {
    pub last_membership: EffectiveMembership<C>,
    ...
}
```

Changed methods:

```rust
RaftStorage::get_membership() -> Result<EffectiveMembership<C>, _>;
RaftStorage::last_applied_state() -> Result<(_, EffectiveMembership<C>), _>;
```

Author: drmingdrmer

example-raft-kv/src/store/mod.rs

@@ -73,7 +73,7 @@ pub struct ExampleStateMachine {
     pub last_applied_log: Option<LogId<ExampleNodeId>>,
 
     // TODO: it should not be Option.
-    pub last_membership: Option<EffectiveMembership<ExampleTypeConfig>>,
+    pub last_membership: EffectiveMembership<ExampleTypeConfig>,
 
     /// Application data.
     pub data: BTreeMap<String, String>,
@@ -252,13 +252,8 @@ impl RaftStorage<ExampleTypeConfig> for Arc<ExampleStore> {
 
     async fn last_applied_state(
         &mut self,
-    ) -> Result<
-        (
-            Option<LogId<ExampleNodeId>>,
-            Option<EffectiveMembership<ExampleTypeConfig>>,
-        ),
-        StorageError<ExampleNodeId>,
-    > {
+    ) -> Result<(Option<LogId<ExampleNodeId>>, EffectiveMembership<ExampleTypeConfig>), StorageError<ExampleNodeId>>
+    {
         let state_machine = self.state_machine.read().await;
         Ok((state_machine.last_applied_log, state_machine.last_membership.clone()))
     }
@@ -288,7 +283,7 @@ impl RaftStorage<ExampleTypeConfig> for Arc<ExampleStore> {
                     }
                 },
                 EntryPayload::Membership(ref mem) => {
-                    sm.last_membership = Some(EffectiveMembership::new(Some(entry.log_id), mem.clone()));
+                    sm.last_membership = EffectiveMembership::new(Some(entry.log_id), mem.clone());
                     res.push(ExampleResponse { value: None })
                 }
             };

memstore/src/lib.rs

@@ -90,7 +90,7 @@ pub struct MemStoreSnapshot {
 pub struct MemStoreStateMachine {
     pub last_applied_log: Option<LogId<MemNodeId>>,
 
-    pub last_membership: Option<EffectiveMembership<Config>>,
+    pub last_membership: EffectiveMembership<Config>,
 
     /// A mapping of client IDs to their state info.
     pub client_serial_responses: HashMap<String, (u64, Option<String>)>,
@@ -263,7 +263,7 @@ impl RaftStorage<Config> for Arc<MemStore> {
 
     async fn last_applied_state(
         &mut self,
-    ) -> Result<(Option<LogId<MemNodeId>>, Option<EffectiveMembership<Config>>), StorageError<MemNodeId>> {
+    ) -> Result<(Option<LogId<MemNodeId>>, EffectiveMembership<Config>), StorageError<MemNodeId>> {
         let sm = self.sm.read().await;
         Ok((sm.last_applied_log, sm.last_membership.clone()))
     }
@@ -343,7 +343,7 @@ impl RaftStorage<Config> for Arc<MemStore> {
                     res.push(ClientResponse(previous));
                 }
                 EntryPayload::Membership(ref mem) => {
-                    sm.last_membership = Some(EffectiveMembership::new(Some(entry.log_id), mem.clone()));
+                    sm.last_membership = EffectiveMembership::new(Some(entry.log_id), mem.clone());
                     res.push(ClientResponse(None))
                 }
             };

openraft/src/core/append_entries.rs

@@ -77,9 +77,6 @@ impl<C: RaftTypeConfig, N: RaftNetworkFactory<C>, S: RaftStorage<C>> RaftCore<C,
         //           initialized. Because a node always commit a membership log as the first log entry.
         let membership = self.storage.get_membership().await?;
 
-        // NOTE: the first log, i.e. log at index 0 can also conflict with the leader and is removed.
-        let membership = membership.unwrap_or_default();
-
         self.update_membership(membership);
 
         tracing::debug!("Done update membership");

openraft/src/core/install_snapshot.rs

@@ -232,10 +232,6 @@ impl<C: RaftTypeConfig, N: RaftNetworkFactory<C>, S: RaftStorage<C>> RaftCore<C,
         let membership = self.storage.get_membership().await?;
         tracing::debug!("storage membership: {:?}", membership);
 
-        assert!(membership.is_some());
-
-        let membership = membership.unwrap();
-
         self.update_membership(membership);
 
         self.snapshot_last_log_id = self.last_applied;

openraft/src/core/mod.rs

@@ -292,7 +292,7 @@ impl<C: RaftTypeConfig, N: RaftNetworkFactory<C>, S: RaftStorage<C>> RaftCore<C,
 
         self.last_log_id = state.last_log_id;
         self.vote = state.vote;
-        self.effective_membership = Arc::new(state.last_membership.unwrap_or_default());
+        self.effective_membership = Arc::new(state.last_membership);
         self.last_applied = state.last_applied;
 
         // NOTE: The commit index must be determined by a leader after

openraft/src/storage.rs

@@ -61,7 +61,7 @@ pub struct InitialState<C: RaftTypeConfig> {
 
     /// The latest cluster membership configuration found, in log or in state machine, else a new initial
     /// membership config consisting only of this node's ID.
-    pub last_membership: Option<EffectiveMembership<C>>,
+    pub last_membership: EffectiveMembership<C>,
 }
 
 /// The state about logs.
@@ -193,18 +193,15 @@ where C: RaftTypeConfig
     /// TODO(xp): if there is no membership log, return `{log_id:None, membership: vec![]}`.
     ///           The raft core should always have an effective_membership.
     ///           Initially, it is empty.
-    async fn get_membership(&mut self) -> Result<Option<EffectiveMembership<C>>, StorageError<C::NodeId>> {
+    async fn get_membership(&mut self) -> Result<EffectiveMembership<C>, StorageError<C::NodeId>> {
         let (_, sm_mem) = self.last_applied_state().await?;
 
-        let sm_mem_next_index = match &sm_mem {
-            None => 0,
-            Some(mem) => mem.log_id.next_index(),
-        };
+        let sm_mem_next_index = sm_mem.log_id.next_index();
 
         let log_mem = self.last_membership_in_log(sm_mem_next_index).await?;
 
-        if log_mem.is_some() {
-            return Ok(log_mem);
+        if let Some(x) = log_mem {
+            return Ok(x);
         }
 
         return Ok(sm_mem);
@@ -311,7 +308,7 @@ where C: RaftTypeConfig
     // NOTE: This can be made into sync, provided all state machines will use atomic read or the like.
     async fn last_applied_state(
         &mut self,
-    ) -> Result<(Option<LogId<C::NodeId>>, Option<EffectiveMembership<C>>), StorageError<C::NodeId>>;
+    ) -> Result<(Option<LogId<C::NodeId>>, EffectiveMembership<C>), StorageError<C::NodeId>>;
 
     /// Apply the given payload of entries to the state machine.
     ///

openraft/src/store_ext.rs

@@ -124,7 +124,7 @@ where
     #[tracing::instrument(level = "trace", skip(self))]
     async fn last_applied_state(
         &mut self,
-    ) -> Result<(Option<LogId<C::NodeId>>, Option<EffectiveMembership<C>>), StorageError<C::NodeId>> {
+    ) -> Result<(Option<LogId<C::NodeId>>, EffectiveMembership<C>), StorageError<C::NodeId>> {
         self.inner().last_applied_state().await
     }
 

openraft/src/testing/suite.rs

@@ -194,7 +194,7 @@ where
 
         let membership = store.get_membership().await?;
 
-        assert!(membership.is_none());
+        assert_eq!(EffectiveMembership::default(), membership);
 
         Ok(())
     }
@@ -218,7 +218,6 @@ where
                 .await?;
 
             let mem = store.get_membership().await?;
-            let mem = mem.unwrap();
 
             assert_eq!(Membership::new(vec![btreeset! {3,4,5}], None), mem.membership,);
         }
@@ -234,8 +233,6 @@ where
 
             let mem = store.get_membership().await?;
 
-            let mem = mem.unwrap();
-
             assert_eq!(Membership::new(vec![btreeset! {3, 4, 5}], None), mem.membership,);
         }
 
@@ -250,8 +247,6 @@ where
 
             let mem = store.get_membership().await?;
 
-            let mem = mem.unwrap();
-
             assert_eq!(Membership::new(vec![btreeset! {7,8,9}], None), mem.membership,);
         }
 
@@ -335,7 +330,7 @@ where
 
             assert_eq!(
                 Membership::new(vec![btreeset! {3,4,5}], None),
-                initial.last_membership.unwrap().membership,
+                initial.last_membership.membership,
             );
         }
 
@@ -352,7 +347,7 @@ where
 
             assert_eq!(
                 Membership::new(vec![btreeset! {3,4,5}], None),
-                initial.last_membership.unwrap().membership,
+                initial.last_membership.membership,
             );
         }
 
@@ -369,7 +364,7 @@ where
 
             assert_eq!(
                 Membership::new(vec![btreeset! {1,2,3}], None),
-                initial.last_membership.unwrap().membership,
+                initial.last_membership.membership,
             );
         }
 
@@ -623,7 +618,7 @@ where
 
         let (applied, membership) = store.last_applied_state().await?;
         assert_eq!(None, applied);
-        assert_eq!(None, membership);
+        assert_eq!(EffectiveMembership::default(), membership);
 
         tracing::info!("--- with last_applied and last_membership");
         {
@@ -637,10 +632,10 @@ where
             let (applied, membership) = store.last_applied_state().await?;
             assert_eq!(Some(LogId::new(LeaderId::new(1, NODE_ID.into()), 3)), applied);
             assert_eq!(
-                Some(EffectiveMembership::new(
+                EffectiveMembership::new(
                     Some(LogId::new(LeaderId::new(1, NODE_ID.into()), 3)),
                     Membership::new(vec![btreeset! {1,2}], None)
-                )),
+                ),
                 membership
             );
         }
@@ -657,10 +652,10 @@ where
             let (applied, membership) = store.last_applied_state().await?;
             assert_eq!(Some(LogId::new(LeaderId::new(1, NODE_ID.into()), 5)), applied);
             assert_eq!(
-                Some(EffectiveMembership::new(
+                EffectiveMembership::new(
                     Some(LogId::new(LeaderId::new(1, NODE_ID.into()), 3)),
                     Membership::new(vec![btreeset! {1,2}], None)
-                )),
+                ),
                 membership
             );
         }

openraft/tests/initialization.rs

@@ -87,10 +87,10 @@ async fn initialization() -> Result<()> {
 
         let sm_mem = sto.last_applied_state().await?.1;
         assert_eq!(
-            Some(EffectiveMembership::new(
+            EffectiveMembership::new(
                 Some(LogId::new(LeaderId::new(0, 0), 0)),
                 Membership::new(vec![btreeset! {0,1,2}], None)
-            )),
+            ),
             sm_mem
         );
     }

openraft/tests/membership/t10_add_learner.rs

@@ -220,7 +220,6 @@ async fn check_learner_after_leader_transfered() -> Result<()> {
     {
         let mut sto = router.get_storage_handle(&1)?;
         let m = sto.get_membership().await?;
-        let m = m.unwrap();
 
         assert_eq!(
             Membership::new(vec![btreeset! {1,3,4}], Some(btreeset! {2})),