Sanity check when unlink_lsn_to_req

Signed-off-by: Xiaoxi Chen <[email protected]>
eBay · Nov 1, 2024 · aed083e · aed083e
1 parent ae19ee5
commit aed083e
Show file tree

Hide file tree

Showing 3 changed files with 13 additions and 6 deletions.
diff --git a/src/lib/replication/repl_dev/raft_repl_dev.cpp b/src/lib/replication/repl_dev/raft_repl_dev.cpp
@@ -895,7 +895,7 @@ void RaftReplDev::handle_commit(repl_req_ptr_t rreq, bool recovery) {
     // Remove the request from repl_key map.
     m_repl_key_req_map.erase(rreq->rkey());
     // Remove the request from lsn map.
-    m_state_machine->unlink_lsn_to_req(rreq->lsn());
+    m_state_machine->unlink_lsn_to_req(rreq->lsn(), rreq);
 
     auto cur_dsn = m_next_dsn.load(std::memory_order_relaxed);
     while (cur_dsn <= rreq->dsn()) {
@@ -1351,7 +1351,7 @@ void RaftReplDev::gc_repl_reqs() {
         // 3. remove from state-machine
         if (removing_rreq->has_state(repl_req_state_t::LOG_FLUSHED)) {
             RD_LOGW("Removing rreq [{}] from state machine, it is risky")
-            m_state_machine->unlink_lsn_to_req(removing_rreq->lsn());
+            m_state_machine->unlink_lsn_to_req(removing_rreq->lsn(), removing_rreq);
         }
     }
 }

diff --git a/src/lib/replication/repl_dev/raft_state_machine.cpp b/src/lib/replication/repl_dev/raft_state_machine.cpp
@@ -219,11 +219,18 @@ uint64_t RaftStateMachine::last_commit_index() {
 
 void RaftStateMachine::become_ready() { m_rd.become_ready(); }
 
-void RaftStateMachine::unlink_lsn_to_req(int64_t lsn) {
+void RaftStateMachine::unlink_lsn_to_req(int64_t lsn, repl_req_ptr_t rreq) {
     auto const it = m_lsn_req_map.find(lsn);
+    // it is possible a LSN mapped to different rreq in history
+    // due to log overwritten. Verify the rreq before removing
     if (it != m_lsn_req_map.cend()) {
-        RD_LOG(DEBUG, "Raft channel: erase lsn {},  rreq {}", lsn, it->second->to_string());
-        m_lsn_req_map.erase(lsn);
+        if (it->second == rreq) {
+            RD_LOG(DEBUG, "Raft channel: erase lsn {},  rreq {}", lsn, it->second->to_string());
+            m_lsn_req_map.erase(lsn);
+        } else {
+            RD_LOGC("Erasing lsn {} pointing to rreq{} differnt with providing rreq {}", lsn, it->second->to_string(),
+                    rreq->to_string());
+        }
     }
 }
 

diff --git a/src/lib/replication/repl_dev/raft_state_machine.h b/src/lib/replication/repl_dev/raft_state_machine.h
@@ -126,7 +126,7 @@ class RaftStateMachine : public nuraft::state_machine {
     repl_req_ptr_t localize_journal_entry_prepare(nuraft::log_entry& lentry);
     repl_req_ptr_t localize_journal_entry_finish(nuraft::log_entry& lentry);
     void link_lsn_to_req(repl_req_ptr_t rreq, int64_t lsn);
-    void unlink_lsn_to_req(int64_t lsn);
+    void unlink_lsn_to_req(int64_t lsn, repl_req_ptr_t rreq);
     repl_req_ptr_t lsn_to_req(int64_t lsn);
     nuraft_mesg::repl_service_ctx* group_msg_service();