add rreq check before appending logs

conanfile.py

@@ -9,7 +9,7 @@
 
 class HomestoreConan(ConanFile):
     name = "homestore"
-    version = "6.20.24"
+    version = "6.20.25"
 
     homepage = "https://github.com/eBay/Homestore"
     description = "HomeStore Storage Engine"

src/lib/replication/log_store/repl_log_store.cpp

@@ -15,8 +15,25 @@ uint64_t ReplLogStore::append(nuraft::ptr< nuraft::log_entry >& entry) {
         return lsn;
     }
 
-    repl_req_ptr_t rreq = m_sm.localize_journal_entry_finish(*entry);
-    RELEASE_ASSERT_NE(nullptr != rreq, "Failed to localize journal entry before appending log");
+    repl_journal_entry* jentry = r_cast< repl_journal_entry const* >(entry->get_buf().data_begin());
+    RELEASE_ASSERT_EQ(jentry->major_version, repl_journal_entry::JOURNAL_ENTRY_MAJOR,
+                      "Mismatched version of journal entry received from RAFT peer");
+
+    repl_key rkey{.server_id = jentry->server_id, .term = entry->get_term(), .dsn = jentry->dsn};
+
+    // make sure the rreq exists before appending the log, both leader and follower
+
+    // 1. for leader, there is still a very corner case that after nuraft::cb_func::Type::ProcessReq check, term changes
+    // and leader appends a log entry with new , and the corresponding rreq is not created yet. This is very rare
+
+    // 2. for follower, when appending a log entry, it must have been proposed by the leader, so the rreq must exist
+    // in memory.
+    auto rreq = m_rd.repl_key_to_req(rkey);
+    if (!m_rd.repl_key_to_req(rkey)) {
+        // in this case, we crash to make sure no data correctness issue happens.
+        RELEASE_ASSERT(false, "Failed to find in-memory rreq for repl_key={} before appending log", rkey.to_string());
+    }
+
     ulong lsn = HomeRaftLogStore::append(entry);
     m_sm.link_lsn_to_req(rreq, int64_cast(lsn));
     RD_LOGT(rreq->traceID(), "Raft Channel: Received append log entry rreq=[{}]", rreq->to_compact_string());

src/lib/replication/repl_dev/raft_repl_dev.cpp

@@ -2057,6 +2057,21 @@ nuraft::cb_func::ReturnCode RaftReplDev::raft_event(nuraft::cb_func::Type type,
         return nuraft::cb_func::ReturnCode::Ok;
     }
 
+    // this is called on leader before appending the log entries.
+    // leader will assign the current term to each log entry, we check if the corrsponding rreq exists. if not , refuse
+    // the request batch.
+    case nuraft::cb_func::Type::ProcessReq: {
+        auto req = r_cast< nuraft::req_msg* >(param->ctx);
+        for (const auto& entry : req->log_entries()) {
+            repl_journal_entry* jentry = r_cast< repl_journal_entry* >(entry->get_buf().data_begin());
+            repl_key rkey{.server_id = jentry->server_id, .term = raft_server()->get_term(), .dsn = jentry->dsn};
+            if (m_repl_key_req_map.find(rkey) == m_repl_key_req_map.end()) {
+                return nuraft::cb_func::ReturnCode::ReturnNull;
+            }
+        }
+        return nuraft::cb_func::ReturnCode::Ok;
+    }
+
     // RemovedFromCluster will be handled in nuraft_mesg::generic_raft_event_handler where leave() is called
 
     // TODO: Add more type handler if necessary

src/lib/replication/repl_dev/raft_state_machine.h

@@ -122,8 +122,10 @@ class RaftStateMachine : public nuraft::state_machine {
 
     ////////// APIs outside of nuraft::state_machine requirements ////////////////////
     ReplServiceError propose_to_raft(repl_req_ptr_t rreq);
+
     repl_req_ptr_t localize_journal_entry_prepare(nuraft::log_entry& lentry, int64_t lsn = -1);
     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, repl_req_ptr_t rreq);
     repl_req_ptr_t lsn_to_req(int64_t lsn);