change: RaftStore::get_log_entries use range as arg; add try_get_log_entry() that does not return error even when defensive check is on

drmingdrmer · drmingdrmer · commit 1c46a71241ad · 2021-08-28T23:07:03.000+08:00
diff --git a/async-raft/src/core/append_entries.rs b/async-raft/src/core/append_entries.rs
@@ -87,11 +87,25 @@ impl<D: AppData, R: AppDataResponse, N: RaftNetwork<D>, S: RaftStorage<D, R>> Ra
         //// Begin Log Consistency Check ////
         tracing::debug!("begin log consistency check");
 
+        if self.last_log_id.index < msg.prev_log_id.index {
+            if report_metrics {
+                self.report_metrics(Update::Ignore);
+            }
+
+            return Ok(AppendEntriesResponse {
+                term: self.current_term,
+                success: false,
+                conflict_opt: Some(ConflictOpt {
+                    log_id: self.last_log_id,
+                }),
+            });
+        }
+
         // Previous log info doesn't immediately line up, so perform log consistency check and proceed based on its
         // result.
         let entries = self
             .storage
-            .get_log_entries(msg.prev_log_id.index, msg.prev_log_id.index + 1)
+            .get_log_entries(msg.prev_log_id.index..=msg.prev_log_id.index)
             .await
             .map_err(|err| self.map_fatal_storage_error(err))?;
         let target_entry = match entries.first() {
@@ -136,7 +150,7 @@ impl<D: AppData, R: AppDataResponse, N: RaftNetwork<D>, S: RaftStorage<D, R>> Ra
             };
             let old_entries = self
                 .storage
-                .get_log_entries(start, msg.prev_log_id.index)
+                .get_log_entries(start..msg.prev_log_id.index)
                 .await
                 .map_err(|err| self.map_fatal_storage_error(err))?;
             let opt = match old_entries.iter().find(|entry| entry.log_id.term == msg.prev_log_id.term) {
@@ -236,7 +250,7 @@ impl<D: AppData, R: AppDataResponse, N: RaftNetwork<D>, S: RaftStorage<D, R>> Ra
         // TODO(xp): logs in storage must be consecutive.
         let entries = self
             .storage
-            .get_log_entries(self.last_applied.index + 1, self.commit_index + 1)
+            .get_log_entries(self.last_applied.index + 1..=self.commit_index)
             .await
             .map_err(|e| self.map_fatal_storage_error(e))?;
 
@@ -290,11 +304,16 @@ impl<D: AppData, R: AppDataResponse, N: RaftNetwork<D>, S: RaftStorage<D, R>> Ra
             return;
         }
 
+        assert!(start <= stop);
+        if start == stop {
+            return;
+        }
+
         // Fetch the series of entries which must be applied to the state machine, then apply them.
         let handle = tokio::spawn(
             async move {
                 let mut new_last_applied: Option<LogId> = None;
-                let entries = storage.get_log_entries(start, stop).await?;
+                let entries = storage.get_log_entries(start..stop).await?;
                 if let Some(entry) = entries.last() {
                     new_last_applied = Some(entry.log_id);
                 }
diff --git a/async-raft/src/core/client.rs b/async-raft/src/core/client.rs
@@ -79,12 +79,8 @@ impl<'a, D: AppData, R: AppDataResponse, N: RaftNetwork<D>, S: RaftStorage<D, R>
             // Thus if a new leader sees only the first one, it needs to append the final config log to let
             // the change-membership operation to finish.
 
-            let last_logs = self
-                .core
-                .storage
-                .get_log_entries(last_index, last_index + 1)
-                .await
-                .map_err(RaftError::RaftStorage)?;
+            let last_logs =
+                self.core.storage.get_log_entries(last_index..=last_index).await.map_err(RaftError::RaftStorage)?;
             let last_log = &last_logs[0];
 
             let req = match last_log.payload {
@@ -418,7 +414,7 @@ impl<'a, D: AppData, R: AppDataResponse, N: RaftNetwork<D>, S: RaftStorage<D, R>
             let entries = self
                 .core
                 .storage
-                .get_log_entries(expected_next_index, index)
+                .get_log_entries(expected_next_index..index)
                 .await
                 .map_err(|err| self.core.map_fatal_storage_error(err))?;
 
diff --git a/async-raft/src/replication/mod.rs b/async-raft/src/replication/mod.rs
@@ -362,16 +362,23 @@ impl<D: AppData, R: AppDataResponse, N: RaftNetwork<D>, S: RaftStorage<D, R>> Re
             }
 
             // Fetch the entry at conflict index and use the term specified there.
-            match self
-                .storage
-                .get_log_entries(conflict.log_id.index, conflict.log_id.index + 1)
-                .await
-                .map(|entries| entries.get(0).map(|entry| entry.log_id.term))
-            {
-                Ok(Some(term)) => {
+            let ent = self.storage.try_get_log_entry(conflict.log_id.index).await;
+            let ent = match ent {
+                Ok(x) => x,
+                Err(err) => {
+                    tracing::error!(error=?err, "error fetching log entry due to returned AppendEntries RPC conflict_opt");
+                    let _ = self.raft_core_tx.send((ReplicaEvent::Shutdown, tracing::debug_span!("CH")));
+                    self.target_state = TargetReplState::Shutdown;
+                    return;
+                }
+            };
+
+            let ent_term = ent.map(|entry| entry.log_id.term);
+            match ent_term {
+                Some(term) => {
                     self.matched.term = term; // If we have the specified log, ensure we use its term.
                 }
-                Ok(None) => {
+                None => {
                     // This condition would only ever be reached if the log has been removed due to
                     // log compaction (barring critical storage failure), so transition to snapshotting.
                     self.target_state = TargetReplState::Snapshotting;
@@ -384,12 +391,6 @@ impl<D: AppData, R: AppDataResponse, N: RaftNetwork<D>, S: RaftStorage<D, R>> Re
                     ));
                     return;
                 }
-                Err(err) => {
-                    tracing::error!(error=%err, "error fetching log entry due to returned AppendEntries RPC conflict_opt");
-                    let _ = self.raft_core_tx.send((ReplicaEvent::Shutdown, tracing::debug_span!("CH")));
-                    self.target_state = TargetReplState::Shutdown;
-                    return;
-                }
             };
 
             // Check snapshot policy and handle conflict as needed.
@@ -676,7 +677,7 @@ impl<'a, D: AppData, R: AppDataResponse, N: RaftNetwork<D>, S: RaftStorage<D, R>
     /// Ensure there are no gaps in the outbound buffer due to transition from lagging.
     #[tracing::instrument(level = "trace", skip(self))]
     async fn frontload_outbound_buffer(&mut self, start: u64, stop: u64) {
-        let entries = match self.replication_core.storage.get_log_entries(start, stop).await {
+        let entries = match self.replication_core.storage.get_log_entries(start..stop).await {
             Ok(entries) => entries,
             Err(err) => {
                 tracing::error!(error=%err, "error while frontloading outbound buffer");
@@ -785,7 +786,7 @@ impl<'a, D: AppData, R: AppDataResponse, N: RaftNetwork<D>, S: RaftStorage<D, R>
             // Bringing the target up-to-date by fetching the largest possible payload of entries
             // from storage within permitted configuration & ensure no snapshot pointer was returned.
             let entries =
-                match self.replication_core.storage.get_log_entries(self.replication_core.next_index, stop_idx).await {
+                match self.replication_core.storage.get_log_entries(self.replication_core.next_index..stop_idx).await {
                     Ok(entries) => entries,
                     Err(err) => {
                         tracing::error!(error=%err, "error fetching logs from storage");
diff --git a/async-raft/src/storage.rs b/async-raft/src/storage.rs
@@ -159,7 +159,14 @@ where
     /// The start value is inclusive in the search and the stop value is non-inclusive: `[start, stop)`.
     ///
     /// Errors returned from this method will cause Raft to go into shutdown.
-    async fn get_log_entries(&self, start: u64, stop: u64) -> Result<Vec<Entry<D>>>;
+    async fn get_log_entries<RNG: RangeBounds<u64> + Clone + Debug + Send + Sync>(
+        &self,
+        range: RNG,
+    ) -> Result<Vec<Entry<D>>>;
+
+    /// Try to get an log entry.
+    /// It does not return an error if in defensive mode and the log entry at `log_index` is not found.
+    async fn try_get_log_entry(&self, log_index: u64) -> Result<Option<Entry<D>>>;
 
     /// Delete all logs in a `range`.
     ///
diff --git a/async-raft/tests/fixtures/mod.rs b/async-raft/tests/fixtures/mod.rs
@@ -555,7 +555,7 @@ impl RaftRouter {
     ) {
         let rt = self.routing_table.read().await;
         for (id, (_node, storage)) in rt.iter() {
-            let last_log = storage.get_log_entries(0, 10_000).await.unwrap().last().unwrap().log_id.index;
+            let last_log = storage.get_log_entries(..).await.unwrap().last().unwrap().log_id.index;
             assert_eq!(
                 last_log, expect_last_log,
                 "expected node {} to have last_log {}, got {}",
diff --git a/async-raft/tests/initialization.rs b/async-raft/tests/initialization.rs
@@ -54,7 +54,7 @@ async fn initialization() -> Result<()> {
 
     for i in 0..3 {
         let sto = router.get_storage_handle(&1).await?;
-        let first = sto.get_log_entries(1, 2).await?.first().cloned();
+        let first = sto.get_log_entries(1..2).await?.first().cloned();
 
         tracing::info!("--- check membership is replicated: id: {}, first log: {:?}", i, first);
         let mem = match first.unwrap().payload {
diff --git a/async-raft/tests/members_leader_fix_partial.rs b/async-raft/tests/members_leader_fix_partial.rs
@@ -71,7 +71,7 @@ async fn members_leader_fix_partial() -> Result<()> {
         )
         .await?;
 
-    let final_log = sto.get_log_entries(want, want + 1).await?[0].clone();
+    let final_log = sto.get_log_entries(want..=want).await?[0].clone();
 
     let m = match final_log.payload {
         EntryPayload::ConfigChange(ref m) => m.membership.clone(),
diff --git a/async-raft/tests/snapshot_uses_prev_snap_membership.rs b/async-raft/tests/snapshot_uses_prev_snap_membership.rs
@@ -74,7 +74,7 @@ async fn snapshot_uses_prev_snap_membership() -> Result<()> {
         router.wait_for_snapshot(&btreeset![0], LogId { term: 1, index: want }, to(), "snapshot").await?;
 
         {
-            let logs = sto0.get_log_entries(0, 1000).await?;
+            let logs = sto0.get_log_entries(..).await?;
             assert_eq!(1, logs.len(), "only one snapshot pointer log");
         }
         let m = sto0.get_membership_config().await?;
@@ -117,7 +117,7 @@ async fn snapshot_uses_prev_snap_membership() -> Result<()> {
     tracing::info!("--- check membership");
     {
         {
-            let logs = sto0.get_log_entries(0, 1000).await?;
+            let logs = sto0.get_log_entries(..).await?;
             assert_eq!(1, logs.len(), "only one snapshot pointer log");
         }
         let m = sto0.get_membership_config().await?;
diff --git a/memstore/src/lib.rs b/memstore/src/lib.rs
@@ -5,6 +5,7 @@ mod test;
 
 use std::cmp::max;
 use std::collections::BTreeMap;
+use std::collections::Bound;
 use std::collections::HashMap;
 use std::fmt::Debug;
 use std::io::Cursor;
@@ -307,18 +308,84 @@ impl MemStore {
         Ok(())
     }
 
-    pub async fn defensive_nonempty_range<RT, RNG: RangeBounds<RT> + Clone + Debug + Send + Iterator>(
+    pub async fn defensive_nonempty_range<RNG: RangeBounds<u64> + Clone + Debug + Send>(
         &self,
         range: RNG,
     ) -> anyhow::Result<()> {
         if !*self.defensive.read().await {
             return Ok(());
         }
-        for _ in range.clone() {
+        let start = match range.start_bound() {
+            Bound::Included(i) => Some(*i),
+            Bound::Excluded(i) => Some(*i + 1),
+            Bound::Unbounded => None,
+        };
+
+        let end = match range.end_bound() {
+            Bound::Included(i) => Some(*i),
+            Bound::Excluded(i) => Some(*i - 1),
+            Bound::Unbounded => None,
+        };
+
+        if start.is_none() || end.is_none() {
+            return Ok(());
+        }
+
+        if start > end {
+            return Err(anyhow::anyhow!("range must be nonempty: {:?}", range));
+        }
+
+        Ok(())
+    }
+
+    pub async fn defensive_range_hits_logs<T: AppData, RNG: RangeBounds<u64> + Debug + Send>(
+        &self,
+        range: RNG,
+        logs: &[Entry<T>],
+    ) -> anyhow::Result<()> {
+        if !*self.defensive.read().await {
             return Ok(());
         }
 
-        Err(anyhow::anyhow!("range must be nonempty: {:?}", range))
+        {
+            let want_first = match range.start_bound() {
+                Bound::Included(i) => Some(*i),
+                Bound::Excluded(i) => Some(*i + 1),
+                Bound::Unbounded => None,
+            };
+
+            let first = logs.first().map(|x| x.log_id.index);
+
+            if want_first.is_some() && first != want_first {
+                return Err(anyhow::anyhow!(
+                    "{:?} want first: {:?}, but {:?}",
+                    range,
+                    want_first,
+                    first
+                ));
+            }
+        }
+
+        {
+            let want_last = match range.end_bound() {
+                Bound::Included(i) => Some(*i),
+                Bound::Excluded(i) => Some(*i - 1),
+                Bound::Unbounded => None,
+            };
+
+            let last = logs.last().map(|x| x.log_id.index);
+
+            if want_last.is_some() && last != want_last {
+                return Err(anyhow::anyhow!(
+                    "{:?} want last: {:?}, but {:?}",
+                    range,
+                    want_last,
+                    last
+                ));
+            }
+        }
+
+        Ok(())
     }
 
     pub async fn defensive_apply_log_id_gt_last<D: AppData>(&self, entries: &[&Entry<D>]) -> anyhow::Result<()> {
@@ -477,14 +544,26 @@ impl RaftStorage<ClientRequest, ClientResponse> for MemStore {
     }
 
     #[tracing::instrument(level = "trace", skip(self))]
-    async fn get_log_entries(&self, start: u64, stop: u64) -> Result<Vec<Entry<ClientRequest>>> {
-        // Invalid request, return empty vec.
-        if start > stop {
-            tracing::error!("get_log_entries: invalid request, start({}) > stop({})", start, stop);
-            return Ok(vec![]);
-        }
+    async fn get_log_entries<RNG: RangeBounds<u64> + Clone + Debug + Send + Sync>(
+        &self,
+        range: RNG,
+    ) -> Result<Vec<Entry<ClientRequest>>> {
+        self.defensive_nonempty_range(range.clone()).await?;
+
+        let res = {
+            let log = self.log.read().await;
+            log.range(range.clone()).map(|(_, val)| val.clone()).collect::<Vec<_>>()
+        };
+
+        self.defensive_range_hits_logs(range, &res).await?;
+
+        Ok(res)
+    }
+
+    #[tracing::instrument(level = "trace", skip(self))]
+    async fn try_get_log_entry(&self, log_index: u64) -> Result<Option<Entry<ClientRequest>>> {
         let log = self.log.read().await;
-        Ok(log.range(start..stop).map(|(_, val)| val.clone()).collect())
+        Ok(log.get(&log_index).cloned())
     }
 
     #[tracing::instrument(level = "trace", skip(self, range), fields(range=?range))]
diff --git a/memstore/src/test.rs b/memstore/src/test.rs

Original file line number	Diff line number	Diff line change
`@@ -71,7 +71,7 @@ async fn members_leader_fix_partial() -> Result<()> {`
`71`	`71`	`)`
`72`	`72`	`.await?;`
`73`	`73`
`74`		`- let final_log = sto.get_log_entries(want, want + 1).await?[0].clone();`
	`74`	`+ let final_log = sto.get_log_entries(want..=want).await?[0].clone();`
`75`	`75`
`76`	`76`	`let m = match final_log.payload {`
`77`	`77`	`EntryPayload::ConfigChange(ref m) => m.membership.clone(),`
Original file line number	Diff line number	Diff line change
`@@ -74,7 +74,7 @@ async fn snapshot_uses_prev_snap_membership() -> Result<()> {`
`74`	`74`	`router.wait_for_snapshot(&btreeset![0], LogId { term: 1, index: want }, to(), "snapshot").await?;`
`75`	`75`
`76`	`76`	`{`
`77`		`- let logs = sto0.get_log_entries(0, 1000).await?;`
	`77`	`+ let logs = sto0.get_log_entries(..).await?;`
`78`	`78`	`assert_eq!(1, logs.len(), "only one snapshot pointer log");`
`79`	`79`	`}`
`80`	`80`	`let m = sto0.get_membership_config().await?;`
`@@ -117,7 +117,7 @@ async fn snapshot_uses_prev_snap_membership() -> Result<()> {`
`117`	`117`	`tracing::info!("--- check membership");`
`118`	`118`	`{`
`119`	`119`	`{`
`120`		`- let logs = sto0.get_log_entries(0, 1000).await?;`
	`120`	`+ let logs = sto0.get_log_entries(..).await?;`
`121`	`121`	`assert_eq!(1, logs.len(), "only one snapshot pointer log");`
`122`	`122`	`}`
`123`	`123`	`let m = sto0.get_membership_config().await?;`