fix consistency check and add test (#140)

* fix consistency check and add test

Signed-off-by: tabokie <xy.tao@outlook.com>

* move failpoint tests

Signed-off-by: tabokie <xy.tao@outlook.com>

* fix CI

Signed-off-by: tabokie <xy.tao@outlook.com>

.github/workflows/rust.yml

@@ -36,7 +36,11 @@ jobs:
       - name: Clippy
         run: cargo clippy --features failpoints --all --all-targets -- -D clippy::all
       - name: Run tests
-        run: RUST_BACKTRACE=1 cargo test --features failpoints --all --verbose -- --nocapture
+        run: |
+          cargo test --all --verbose
+          cargo test --test failpoints --features failpoints --verbose -- --test-threads 1 --nocapture
+        env:
+          RUST_BACKTRACE: 1
       - name: Run asan tests
         if: ${{ matrix.os == 'ubuntu-latest' }}
         run: cargo test -Zbuild-std --target x86_64-unknown-linux-gnu --features failpoints --verbose -- --nocapture
@@ -62,7 +66,9 @@ jobs:
       - name: Install grcov
         run: if [[ ! -e ~/.cargo/bin/grcov ]]; then cargo install --locked grcov; fi
       - name: Run tests
-        run: cargo test --features failpoints --all --verbose
+        run: |
+          cargo test --all --verbose
+          cargo test --test failpoints --features failpoints -- --test-threads 1 --nocapture
         env:
           RUSTFLAGS: '-Zinstrument-coverage'
           LLVM_PROFILE_FILE: '%p-%m.profraw'

Cargo.toml

@@ -8,6 +8,17 @@ edition = "2018"
 name = "append-compact-purge"
 path = "examples/append_compact_purge.rs"
 
+[[test]]
+name = "failpoints"
+path = "tests/failpoints/mod.rs"
+required-features = ["failpoints"]
+
+[[bench]]
+name = "benches"
+path = "tests/benches/mod.rs"
+harness = false
+required-features = ["failpoints"]
+
 [dependencies]
 byteorder = "1.2"
 crc32fast = "1.2"
@@ -53,8 +64,3 @@ protobuf-codegen = { git = "https://github.com/pingcap/rust-protobuf", rev = "82
 
 [workspace]
 members = [ "stress" ]
-
-[[bench]]
-name = "bench_recovery"
-harness = false
-required-features = ["failpoints"]

README.md

@@ -67,7 +67,8 @@ Contributions are always welcome! Here are a few tips for making a PR:
 ```
 cargo fmt --all -- --check
 cargo clippy --all --all-targets -- -D clippy::all
-cargo test --features failpoints --all
+cargo test --all
+cargo test --test failpoints --features failpoints -- --test-threads 1
 ```
 
 - For changes that might induce performance effects, please quote the targeted benchmark results in the PR description. In addition to micro-benchmarks, there is a standalone [stress test tool](https://github.com/tikv/raft-engine/tree/master/stress) which you can use to demonstrate the system performance.

src/consistency.rs

@@ -9,15 +9,15 @@ use crate::Result;
 
 #[derive(Default)]
 pub struct ConsistencyChecker {
-    // Raft group id -> last index
-    pending: HashMap<u64, u64>,
-    // Raft group id -> last uncorrupted index
+    // Raft group id -> first index, last index
+    raft_groups: HashMap<u64, (u64, u64)>,
+    // Raft group id -> last unaffected index
     corrupted: HashMap<u64, u64>,
 }
 
 impl ConsistencyChecker {
-    pub fn finish(self) -> Vec<(u64, u64)> {
-        self.corrupted.into_iter().collect()
+    pub fn finish(self) -> HashMap<u64, u64> {
+        self.corrupted
     }
 }
 
@@ -29,26 +29,40 @@ impl ReplayMachine for ConsistencyChecker {
                     let incoming_first_index = ents.0.first().unwrap().index;
                     let incoming_last_index = ents.0.last().unwrap().index;
                     let last_index = self
-                        .pending
+                        .raft_groups
                         .entry(item.raft_group_id)
-                        .or_insert(incoming_last_index);
-                    if *last_index + 1 < incoming_first_index {
+                        .or_insert((incoming_first_index, incoming_last_index));
+                    if last_index.1 + 1 < incoming_first_index {
                         self.corrupted
-                            .insert(item.raft_group_id, incoming_first_index);
+                            .entry(item.raft_group_id)
+                            .or_insert(last_index.1);
                     }
-                    *last_index = incoming_last_index;
+                    last_index.1 = incoming_last_index;
                 }
             }
         }
         Ok(())
     }
 
     fn merge(&mut self, mut rhs: Self, _queue: LogQueue) -> Result<()> {
-        for (id, last_index) in rhs.pending.drain() {
-            self.pending.insert(id, last_index);
+        let mut new_corrupted: HashMap<u64, u64> = HashMap::default();
+        // Find holes between self and rhs.
+        for (id, (first, last)) in rhs.raft_groups.drain() {
+            self.raft_groups
+                .entry(id)
+                .and_modify(|(_, l)| {
+                    if *l + 1 < first {
+                        new_corrupted.insert(id, *l);
+                    }
+                    *l = last;
+                })
+                .or_insert((first, last));
+        }
+        for (id, last_index) in new_corrupted.drain() {
+            self.corrupted.entry(id).or_insert(last_index);
         }
         for (id, last_index) in rhs.corrupted.drain() {
-            self.corrupted.insert(id, last_index);
+            self.corrupted.entry(id).or_insert(last_index);
         }
         Ok(())
     }