Repository: tikv/tikv (language: rust)

Issue:
TiKV panic with "assertion failed: cached_last < trunc_to_idx"
## Bug Report

```log
store/TiDB master bank-multitable auto-retry auto-retry-limit :default update-in-place select FOR UPDATE txn-mode pessimistic isolation :repeatable-read nemesis partition-half,partition-pd-leader,partition-ring,schedules/20241119T123842.000Z/node-1.node-peer.jepsen-tps-7527309-1-659/kv.log:[2024/11/19 12:48:10.846 +00:00] [FATAL] [lib.rs:480] ["assertion failed: cached_last < trunc_to_idx"] [backtrace="   0: tikv_util::set_panic_hook::{{closure}}\             at /home/jenkins/agent/workspace/build-common/go/src/github.com/pingcap/tikv/components/tikv_util/src/lib.rs:479:18
   1: <alloc::boxed::Box<F,A> as core::ops::function::Fn<Args>>::call
             at /rust/toolchains/nightly-2023-12-28-x86_64-unknown-linux-gnu/lib/rustlib/src/rust/library/alloc/src/boxed.rs:2029:9
      std::panicking::rust_panic_with_hook
             at /rust/toolchains/nightly-2023-12-28-x86_64-unknown-linux-gnu/lib/rustlib/src/rust/library/std/src/panicking.rs:783:13
   2: std::panicking::begin_panic_handler::{{closure}}
             at /rust/toolchains/nightly-2023-12-28-x86_64-unknown-linux-gnu/lib/rustlib/src/rust/library/std/src/panicking.rs:649:13
   3: std::sys_common::backtrace::__rust_end_short_backtrace
             at /rust/toolchains/nightly-2023-12-28-x86_64-unknown-linux-gnu/lib/rustlib/src/rust/library/std/src/sys_common/backtrace.rs:171:18
   4: rust_begin_unwind
             at /rust/toolchains/nightly-2023-12-28-x86_64-unknown-linux-gnu/lib/rustlib/src/rust/library/std/src/panicking.rs:645:5
   5: core::panicking::panic_fmt
             at /rust/toolchains/nightly-2023-12-28-x86_64-unknown-linux-gnu/lib/rustlib/src/rust/library/core/src/panicking.rs:72:14
   6: core::panicking::panic
             at /rust/toolchains/nightly-2023-12-28-x86_64-unknown-linux-gnu/lib/rustlib/src/rust/library/core/src/panicking.rs:144:5
   7: raftstore::store::entry_storage::EntryCache::append_impl
             at /home/jenkins/agent/workspace/build-common/go/src/github.com/pingcap/tikv/components/raftstore/src/store/entry_storage.rs:202:21
      raftstore::store::entry_storage::EntryCache::append
             at /home/jenkins/agent/workspace/build-common/go/src/github.com/pingcap/tikv/components/raftstore/src/store/entry_storage.rs:156:32
   8: raftstore::store::entry_storage::EntryStorage<EK,ER>::append
             at /home/jenkins/agent/workspace/build-common/go/src/github.com/pingcap/tikv/components/raftstore/src/store/entry_storage.rs:1098:9
   9: raftstore::store::peer_storage::PeerStorage<EK,ER>::handle_raft_ready
             at /home/jenkins/agent/workspace/build-common/go/src/github.com/pingcap/tikv/components/raftstore/src/store/peer_storage.rs:993:13
      raftstore::store::peer::Peer<EK,ER>::handle_raft_ready_append
             at /home/jenkins/agent/workspace/build-common/go/src/github.com/pingcap/tikv/components/raftstore/src/store/peer.rs:2878:37
  10: raftstore::store::fsm::peer::PeerFsmDelegate<EK,ER,T>::collect_ready
             at /home/jenkins/agent/workspace/build-common/go/src/github.com/pingcap/tikv/components/raftstore/src/store/fsm/peer.rs:2173:19
      <raftstore::store::fsm::store::RaftPoller<EK,ER,T> as batch_system::batch::PollHandler<raftstore::store::fsm::peer::PeerFsm<EK,ER>,raftstore::store::fsm::store::StoreFsm<EK>>>::handle_normal
             at /home/jenkins/agent/workspace/build-common/go/src/github.com/pingcap/tikv/components/raftstore/src/store/fsm/store.rs:1101:13
  11: batch_system::batch::Poller<N,C,Handler>::poll
             at /home/jenkins/agent/workspace/build-common/go/src/github.com/pingcap/tikv/components/batch-system/src/batch.rs:416:27
  12: batch_system::batch::BatchSystem<N,C>::start_poller::{{closure}}
             at /home/jenkins/agent/workspace/build-common/go/src/github.com/pingcap/tikv/components/batch-system/src/batch.rs:589:17
      <std::thread::Builder as tikv_util::sys::thread::StdThreadBuildWrapper>::spawn_wrapper::{{closure}}
             at /home/jenkins/agent/workspace/build-common/go/src/github.com/pingcap/tikv/components/tikv_util/src/sys/thread.rs:438:13
      std::sys_common::backtrace::__rust_begin_short_backtrace
             at /rust/toolchains/nightly-2023-12-28-x86_64-unknown-linux-gnu/lib/rustlib/src/rust/library/std/src/sys_common/backtrace.rs:155:18
  13: std::thread::Builder::spawn_unchecked_::{{closure}}::{{closure}}
             at /rust/toolchains/nightly-2023-12-28-x86_64-unknown-linux-gnu/lib/rustlib/src/rust/library/std/src/thread/mod.rs:529:17
      <core::panic::unwind_safe::AssertUnwindSafe<F> as core::ops::function::FnOnce<()>>::call_once
             at /rust/toolchains/nightly-2023-12-28-x86_64-unknown-linux-gnu/lib/rustlib/src/rust/library/core/src/panic/unwind_safe.rs:272:9
      std::panicking::try::do_call
             at /rust/toolchains/nightly-2023-12-28-x86_64-unknown-linux-gnu/lib/rustlib/src/rust/library/std/src/panicking.rs:552:40
      std::panicking::try
             at /rust/toolchains/nightly-2023-12-28-x86_64-unknown-linux-gnu/lib/rustlib/src/rust/library/std/src/panicking.rs:516:19
      std::panic::catch_unwind
             at /rust/toolchains/nightly-2023-12-28-x86_64-unknown-linux-gnu/lib/rustlib/src/rust/library/std/src/panic.rs:142:14
      std::thread::Builder::spawn_unchecked_::{{closure}}
             at /rust/toolchains/nightly-2023-12-28-x86_64-unknown-linux-gnu/lib/rustlib/src/rust/library/std/src/thread/mod.rs:528:30
      core::ops::function::FnOnce::call_once{{vtable.shim}}
             at /rust/toolchains/nightly-2023-12-28-x86_64-unknown-linux-gnu/lib/rustlib/src/rust/library/core/src/ops/function.rs:250:5
  14: <alloc::boxed::Box<F,A> as core::ops::function::FnOnce<Args>>::call_once
             at /rust/toolchains/nightly-2023-12-28-x86_64-unknown-linux-gnu/lib/rustlib/src/rust/library/alloc/src/boxed.rs:2015:9
      <alloc::boxed::Box<F,A> as core::ops::function::FnOnce<Args>>::call_once
             at /rust/toolchains/nightly-2023-12-28-x86_64-unknown-linux-gnu/lib/rustlib/src/rust/library/alloc/src/boxed.rs:2015:9
      std::sys::unix::thread::Thread::new::thread_start
             at /rust/toolchains/nightly-2023-12-28-x86_64-unknown-linux-gnu/lib/rustlib/src/rust/library/std/src/sys/unix/thread.rs:108:17
  15: start_thread
  16: clone
"] [location=components/raftstore/src/store/entry_storage.rs:202] [thread_name=raftstore-1-1] [thread_id=144]
```

https://tcms.pingcap.net/dashboard/executions/plan/7527309

This panic is found by @SpadeA-Tang , thank you!

### What version of TiKV are you using?
<!-- You can run `tikv-server --version` -->

Nightly 2024-11-18

### Steps to reproduce
<!-- If possible, provide a recipe for reproducing the error. A complete runnable program is good. -->

Run jepsen-bank-tbls-pessimistic



Discussion / hints:
Briefly look into this issue, seems the panic in this issue actually means **A raft peer should not receive append raft entries 
again if they are already committed.** (The panic is caused by L202.)

https://github.com/tikv/tikv/blob/3bd8c24d1a1eb277877557b35d3b175330e7010e/components/raftstore/src/store/entry_storage.rs#L183-L210

But I think this is a too strong assertion that the raft protocal does not requires. The raft protocal only requires that if an entry is committed, the entry with this index will never change. It does not forbid a raft peer to receive some raft entries again even if the peer knows these entries are already committed iff these raft entries are not changed.

I think in the implementation of raftstore, one possible scenario that can triggers this assert failure is that: If an AppendMsg is lost in the network which will triggers append reject after a follower receives the next ApplyMsg and it will response with its current committed index which can be smaller the this leader's committed index as well as itself's last index. Then the leader will resend raft logs from this committed index. But before the next AppendMsg is send(from the follower's committed index), the leader may have sent a Heartbeat msg that advanced the follower's committed index and triggered apply. Then, when the follower receives the AppendMsg from its pervious committed index, this assert will fail.

/cc @BusyJay @gengliqi @overvenus @Connor1996 What do you think?
raft-rs will find the term/index conflict and ensure entry with same index and term will be only appended once. Panic in cache probably means bug in application (like cache index error).
> raft-rs will find the term/index conflict and ensure entry with same index and term will be only appended once. Panic in cache probably means bug in application (like cache index error).

Yes, after re-reading raft-rs, I found raft-rs will directly reject Append Message if the message index is smaller than its current committed index, so what I said above is not possible.
From the code I can imagine one possible cause of the panic, which is when there is a hole in the entry cache. For example:
- Entry cache state: [(log_index=2), (log_index=4), (log_index=5)]
- Commit index: 2
- Index of new entry (`first_index`): 3

When adding the new entry to the entry cache, truncation will occur. `truncate_to` will be set to 0 and `trunc_to_idx` will be set to 2, which overlaps with the committed entry at log_index=2, violating the assertion.

I suspect this might be related to the cache warm-up process but I haven't found anything conclusive yet.

Another reproduction: https://tcms.pingcap.net/dashboard/executions/case/13272927
It was tested based on #17806 and contains the debug message #18002 . 
Another reproduction: https://tcms.pingcap.net/dashboard/executions/case/13355965
Another reproduction: https://tcms.pingcap.net/dashboard/executions/plan/7818989
The tikv is built from 8a68c38 @ #17806 .
This issue is triggered in the following scenario: A isolated leader (has uncommitted raft entries in its entry_cache) receives a new batch of raft messages for a new leader after it reconnecting to other peers. If this batch, if the committed index is high that its current committed index, due to the feature "apply committed raft logs before persist",  in the following `handle_raft_ready_append`, the committed raft message will not be empty and then will be sent to the ApplyFsm, and the when it try to append new logs to the EntryCache, it will triggers the panic because the newly committed logs(sending to apply fsm)' index is higher.

Here is a example to make it easier to understand:
1. Say there is a raft group with A, B, C, and A is the current leader. All three peer's persist, commit and applied index are 10.
2. A is isolated from B and C. There comes a new Propose with index 11. This new raft log can't be committed due to isolation and it will be put in the EntryCache.
3. After an election timeout, a new leader will be elected, let's say it is B. 
4. There come a new Propose to B with index 11, and it is committed in B and C.
5. after some time, A's isolation is fixed. If at this time, A still think it is the leader, and it receives a new RaftMessage batch from B(index = 11) and committed index also with 11. After handle this batch, A will become follower with committed index 11. So at `handle_raft_ready`, in handle_raft_committed_entries, it is possible to send the raft messge(index 11) to apply fsm, and then in `PeerStore::handle_raft_ready`, it want to put the new entry(index = 11) to entry store, this panic will occur because it tries to truncate the old entry(index = 11) but the the the entry(index = 11) is applying.

Further analysis:
1. From the example we can see in step 5, we want to apply unpersisted raft entry(index = 11) on raft follower which is unexpected. This can happen because we reset the `max_apply_unpersisted_log_limit` in the `on_role_change` but collect committed entries in `collect_ready`. Because `collect_ready` happens before `on_role_change`, so in the same round, the new follower can apply more raft entries than we expected.
2. Let's talk about the `assertion failed: cached_last < trunc_to_idx` again. Originally, it purpose is to ensure that committed entries should not be changed. But in our implement, we handle the events in the following order: collect_ready --> handle_apply(committed entries) --> write to entry cache. Thus, this assertion is no longer correct after we support the feature "apply committed raft logs before persist".

TODO:
- Due to analysis 1, we should change raft-rs to only allow return committed but not persisted entries on leader.
- Remove this assertion as it is no longer true.

BTW, I found it very hard to write a test case that can reproduce this scenario, @overvenus @hbisheng @Connor1996 , do you have any ideas?


Fix patch:
diff --git a/Cargo.lock b/Cargo.lock
index 89d3220e3b..61d57769d7 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -5179,7 +5179,7 @@ dependencies = [
 [[package]]
 name = "raft"
 version = "0.7.0"
-source = "git+https://github.com/tikv/raft-rs?branch=master#a76fb6ef2cbd002ec10d63a2ac68b4a20b20fe3e"
+source = "git+https://github.com/tikv/raft-rs?branch=master#0d01b20312f74889a5e44ad4180aade5da2f16fa"
 dependencies = [
  "bytes",
  "fxhash",
@@ -5238,7 +5238,7 @@ dependencies = [
 [[package]]
 name = "raft-proto"
 version = "0.7.0"
-source = "git+https://github.com/tikv/raft-rs?branch=master#a76fb6ef2cbd002ec10d63a2ac68b4a20b20fe3e"
+source = "git+https://github.com/tikv/raft-rs?branch=master#0d01b20312f74889a5e44ad4180aade5da2f16fa"
 dependencies = [
  "bytes",
  "protobuf",
diff --git a/components/raftstore/src/store/entry_storage.rs b/components/raftstore/src/store/entry_storage.rs
index 5db6916c75..66e2197902 100644
--- a/components/raftstore/src/store/entry_storage.rs
+++ b/components/raftstore/src/store/entry_storage.rs
@@ -199,7 +199,14 @@ impl EntryCache {
                     // Only committed entries can be traced, and only uncommitted entries
                     // can be truncated. So there won't be any overlaps.
                     let cached_last = cached.range.end - 1;
-                    assert!(cached_last < trunc_to_idx);
+                    // TODO: if we are going to support apply unpersisted entries on
+                    // follower, this assert will not longer be correct.
+                    assert!(
+                        cached_last < trunc_to_idx,
+                        "cached_last: {}, trunc_to_idx: {}",
+                        cached_last,
+                        trunc_to_idx
+                    );
                 }
             } else if cache_last_index + 1 < first_index {
                 panic!(
diff --git a/components/raftstore/src/store/peer.rs b/components/raftstore/src/store/peer.rs
index 54c3bbac73..6daa7bef62 100644
--- a/components/raftstore/src/store/peer.rs
+++ b/components/raftstore/src/store/peer.rs
@@ -806,6 +806,12 @@ where
     /// The index of last scheduled committed raft log.
     pub last_applying_idx: u64,
     pub max_apply_unpersisted_log_limit: u64,
+    /// A flag used to track whether `max_apply_unpersisted_log_limit` is set
+    /// to the Peer in raft-rs. We need this flag to handle the metrics
+    /// `RAFT_ENABLE_UNPERSISTED_APPLY_GAUGE` correctly because raft-rs can
+    /// reset `max_apply_unpersisted_log_limit` to 0 when it demotes from
+    /// leader.
+    enable_apply_unpersisted_log_state: bool,
     /// The minimum raft index after which apply unpersisted raft log can be
     /// enabled. We force disable apply unpersisted raft log in following 2
     /// situation:
@@ -1036,6 +1042,7 @@ where
             tag: tag.clone(),
             last_applying_idx: applied_index,
             max_apply_unpersisted_log_limit: cfg.max_apply_unpersisted_log_limit,
+            enable_apply_unpersisted_log_state: false,
             min_safe_index_for_unpersisted_apply: last_index,
             last_compacted_idx: 0,
             last_compacted_time: Instant::now(),
@@ -1217,6 +1224,7 @@ where
                     == 0
             {
                 RAFT_ENABLE_UNPERSISTED_APPLY_GAUGE.inc();
+                self.enable_apply_unpersisted_log_state = true;
             }
             self.raft_group
                 .raft
@@ -1237,6 +1245,12 @@ where
             > 0
         {
             self.raft_group.raft.set_max_apply_unpersisted_log_limit(0);
+        }
+        // NOTE: `max_apply_unpersisted_log_limit` can be reset in raft-rs when leader
+        // demote to follower, in this case, we should still decrease the
+        // metrics counter.
+        if self.enable_apply_unpersisted_log_state {
+            self.enable_apply_unpersisted_log_state = false;
             RAFT_ENABLE_UNPERSISTED_APPLY_GAUGE.dec();
         }
     }


Test patch:
diff --git a/tests/failpoints/cases/test_early_apply.rs b/tests/failpoints/cases/test_early_apply.rs
index 540ce408d5..c2e8cb7960 100644
--- a/tests/failpoints/cases/test_early_apply.rs
+++ b/tests/failpoints/cases/test_early_apply.rs
@@ -1,13 +1,20 @@
 // Copyright 2020 TiKV Project Authors. Licensed under Apache-2.0.
 
-use std::sync::{
-    atomic::{AtomicBool, Ordering},
-    Arc,
+use std::{
+    sync::{
+        atomic::{AtomicBool, Ordering},
+        Arc, Mutex,
+    },
+    time::Duration,
 };
 
+use kvproto::raft_serverpb::RaftMessage;
+use pd_client::PdClient;
 use raft::eraftpb::MessageType;
+use raftstore::store::{InspectedRaftMessage, PeerMsg};
 use test_raftstore::*;
 use test_raftstore_macro::test_case;
+use tikv_util::future::block_on_timeout;
 
 // Test if a singleton can apply a log before persisting it.
 #[test_case(test_raftstore::new_node_cluster)]
@@ -159,3 +166,87 @@ fn test_early_apply_yield_followed_with_many_entries() {
 
     must_get_equal(&cluster.get_engine(3), b"k150", b"v150");
 }
+
+// Test the consistency of EntryCache when partitioned leader contains
+// uncommitted propose, and after the partition is recovered, it can replicate
+// raft entries for new leader correctly. This case tests the corner scenario
+// that partitioned leader receives a new Append msg from new elected leader and
+// the new entries are already committed and overlap with existing
+// uncommitted entries in the entry cache, it may cause panic if handled
+// incorrectly. See issue https://github.com/tikv/tikv/issues/17868 for more details.
+#[test]
+fn test_early_apply_leader_demote_by_append() {
+    let mut cluster = new_node_cluster(0, 3);
+    cluster.pd_client.disable_default_operator();
+
+    cluster.run();
+
+    cluster.must_put(b"k1", b"v1");
+    let region = cluster.pd_client.get_region(b"k1").unwrap();
+    cluster.must_transfer_leader(region.id, new_peer(1, 1));
+
+    // isolate 1 for 2,3
+    let dropped_append: Arc<Mutex<RaftMessage>> = Arc::new(Mutex::new(RaftMessage::default()));
+    let msg_ref = dropped_append.clone();
+    cluster.add_recv_filter_on_node(
+        1,
+        Box::new(DropMessageFilter::new(Arc::new(move |m| {
+            // save dropped append.
+            if m.get_message().msg_type == MessageType::MsgAppend {
+                *msg_ref.lock().unwrap() = m.clone();
+            }
+
+            false
+        }))),
+    );
+    for id in [2, 3] {
+        cluster.add_recv_filter_on_node(
+            id,
+            Box::new(DropMessageFilter::new(Arc::new(|m| {
+                m.get_message().from != 1
+            }))),
+        );
+    }
+
+    // propose a new write, the write should timeout.
+    let ch = cluster.async_put(b"k1", b"v2").unwrap();
+    block_on_timeout(ch, Duration::from_millis(100)).unwrap_err();
+
+    let fp = "pause_on_peer_collect_message";
+    // pause peer 1 to wait for leader timeout.
+    fail::cfg(fp, "pause").unwrap();
+
+    // wait for leader timeout.
+    sleep_ms(200);
+
+    cluster.reset_leader_of_region(region.id);
+    let leader = cluster.leader_of_region(region.id).unwrap();
+    assert_ne!(leader.store_id, 1);
+
+    cluster.must_put(b"k1", b"v3");
+
+    // Send the dropped msg to store 1.
+    let mut msg = dropped_append.lock().unwrap().clone();
+    assert_eq!(msg.get_message().to, 1);
+    // Advance the committed index of the Append msg to trigger the corner case.
+    // I don't find an easy way to trigger this kind of msg, so direct modify the
+    // first append to mock that scenario.
+    let entry_idx = msg
+        .get_message()
+        .get_entries()
+        .last()
+        .map(|e| e.index)
+        .unwrap();
+    msg.mut_message().commit = entry_idx;
+    let peer_msg = PeerMsg::RaftMessage(Box::new(InspectedRaftMessage { heap_size: 0, msg }), None);
+    cluster.get_router(1).unwrap().send(1, peer_msg).unwrap();
+
+    for i in 1..=3 {
+        cluster.clear_recv_filter_on_node(i);
+    }
+
+    // remove fp.
+    fail::remove(fp);
+    cluster.must_put(b"k1", b"v4");
+    eventually_get_equal(&cluster.get_engine(1), b"k1", b"v4");
+}
diff --git a/tests/failpoints/cases/test_split_region.rs b/tests/failpoints/cases/test_split_region.rs
index 916258a354..59db7ad88b 100644
--- a/tests/failpoints/cases/test_split_region.rs
+++ b/tests/failpoints/cases/test_split_region.rs
@@ -75,7 +75,7 @@ fn test_meta_inconsistency() {
     let region = cluster.get_region(b"");
     cluster.must_split(&region, b"k5");
 
-    // Scheduler a larger peed id heartbeat msg to trigger peer destroy for peer
+    // Scheduler a larger peer id heartbeat msg to trigger peer destroy for peer
     // 1003, pause it before the meta.lock operation so new region insertions by
     // region split could go first.
     // Thus a inconsistency could happen because the destroy is handled
@@ -722,7 +722,7 @@ fn test_split_continue_when_destroy_peer_after_mem_check() {
     })
     .unwrap();
 
-    // Resum region 1000 processing and wait till it's destroyed.
+    // Resume region 1000 processing and wait till it's destroyed.
     fail::remove(before_check_snapshot_1000_2_fp);
     destroy_rx.recv_timeout(Duration::from_secs(3)).unwrap();