fix init rreq

conanfile.py

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

src/include/homestore/replication/repl_dev.h

@@ -72,7 +72,8 @@ struct repl_key {
 
     bool operator==(repl_key const& other) const = default;
     std::string to_string() const {
-        return fmt::format("server={}, term={}, dsn={}, hash={}", server_id, term, dsn, Hasher()(*this));
+        return fmt::format("server={}, term={}, dsn={}, hash={}, trace_id={}", server_id, term, dsn, Hasher()(*this),
+                           traceID);
     }
 };
 

src/lib/replication/log_store/repl_log_store.cpp

@@ -15,7 +15,7 @@ uint64_t ReplLogStore::append(nuraft::ptr< nuraft::log_entry >& entry) {
         return lsn;
     }
 
-    repl_req_ptr_t rreq = m_sm.localize_journal_entry_finish(*entry, true /* is_append_log */);
+    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");
 
     ulong lsn = HomeRaftLogStore::append(entry);

src/lib/replication/repl_dev/raft_repl_dev.cpp

@@ -260,10 +260,13 @@ AsyncReplResult<> RaftReplDev::start_replace_member(std::string& task_id, const
 
     if (status != ReplServiceError::OK) {
         RD_LOGE(trace_id, "Initializing rreq failed, rreq=[{}], error={}", rreq->to_string(), status);
+        reset_quorum_size(0, trace_id);
+        decr_pending_request_num();
+        return make_async_error<>(ReplServiceError::RETRY_REQUEST);
     }
 
-    auto const [_, happened] = m_repl_key_req_map.emplace(rreq->rkey(), rreq);
-    RD_DBG_ASSERT(happened, "Duplicate repl_key={} found in the map", rreq->rkey().to_string());
+    // the rreq will be recreated when appending log, which is a follower-like flow
+    // TODO: refactor the rreq creation, we should put it into m_repl_key_req_map before we call propose_to_raft.
 
     auto err = m_state_machine->propose_to_raft(std::move(rreq));
     if (err != ReplServiceError::OK) {
@@ -365,8 +368,8 @@ AsyncReplResult<> RaftReplDev::complete_replace_member(std::string& task_id, con
 
     // Step 2. Append log entry to complete replace member
     RD_LOGI(trace_id,
-            "Step6. Replace member, propose to raft for HS_CTRL_COMPLETE_REPLACE req, group_id={}, task_id={}", task_id,
-            group_id_str());
+            "Step6. Replace member, propose to raft for HS_CTRL_COMPLETE_REPLACE req, group_id={}, task_id={}",
+            group_id_str(), task_id);
     auto rreq = repl_req_ptr_t(new repl_req_ctx{});
     auto ctx = replace_member_ctx(task_id, member_out, member_in);
 
@@ -382,10 +385,13 @@ AsyncReplResult<> RaftReplDev::complete_replace_member(std::string& task_id, con
 
     if (status != ReplServiceError::OK) {
         RD_LOGE(trace_id, "Initializing rreq failed, rreq=[{}], error={}", rreq->to_string(), status);
+        reset_quorum_size(0, trace_id);
+        decr_pending_request_num();
+        return make_async_error<>(ReplServiceError::RETRY_REQUEST);
     }
 
-    auto const [_, happened] = m_repl_key_req_map.emplace(rreq->rkey(), rreq);
-    RD_DBG_ASSERT(happened, "Duplicate repl_key={} found in the map", rreq->rkey().to_string());
+    // the rreq will be recreated when appending log, which is a follower-like flow
+    // TODO: refactor the rreq creation, we should put it into m_repl_key_req_map before we call propose_to_raft.
 
     auto err = m_state_machine->propose_to_raft(std::move(rreq));
     if (err != ReplServiceError::OK) {
@@ -701,8 +707,8 @@ folly::SemiFuture< ReplServiceError > RaftReplDev::destroy_group() {
         return folly::makeSemiFuture< ReplServiceError >(std::move(err));
     }
 
-    auto const [_, happened] = m_repl_key_req_map.emplace(rreq->rkey(), rreq);
-    RD_DBG_ASSERT(happened, "Duplicate repl_key={} found in the map", rreq->rkey().to_string());
+    // the rreq will be recreated when appending log, which is a follower-like flow
+    // TODO: refactor the rreq creation, we should put it into m_repl_key_req_map before we call propose_to_raft.
 
     err = m_state_machine->propose_to_raft(std::move(rreq));
     if (err != ReplServiceError::OK) {
@@ -764,10 +770,12 @@ void RaftReplDev::propose_truncate_boundary() {
         if (status != ReplServiceError::OK) {
             RD_LOGE(std::numeric_limits< uint64_t >::max(), "Initializing rreq failed, rreq=[{}], error={}",
                     rreq->to_string(), status);
+            decr_pending_request_num();
+            return;
         }
 
-        auto const [_, happened] = m_repl_key_req_map.emplace(rreq->rkey(), rreq);
-        RD_DBG_ASSERT(happened, "Duplicate repl_key={} found in the map", rreq->rkey().to_string());
+        // the rreq will be recreated when appending log, which is a follower-like flow
+        // TODO: refactor the rreq creation, we should put it into m_repl_key_req_map before we call propose_to_raft.
 
         auto err = m_state_machine->propose_to_raft(std::move(rreq));
         if (err != ReplServiceError::OK) {
@@ -2171,9 +2179,8 @@ void RaftReplDev::monitor_replace_member_replication_status() {
 
     trace_id_t trace_id = generateRandomTraceId();
 
-    RD_LOGD(trace_id, "Trigger complete_replace_member, task_id={}, replica_in={}, replica_out={}",
-            boost::uuids::to_string(replica_in), boost::uuids::to_string(replica_in),
-            boost::uuids::to_string(replica_out));
+    RD_LOGD(trace_id, "Trigger complete_replace_member, task_id={}, replica_in={}, replica_out={}", task_id,
+            boost::uuids::to_string(replica_in), boost::uuids::to_string(replica_out));
 
     replica_member_info out{replica_out, ""};
     replica_member_info in{replica_in, ""};

src/lib/replication/repl_dev/raft_state_machine.cpp

@@ -133,7 +133,7 @@ repl_req_ptr_t RaftStateMachine::localize_journal_entry_prepare(nuraft::log_entr
     return rreq;
 }
 
-repl_req_ptr_t RaftStateMachine::localize_journal_entry_finish(nuraft::log_entry& lentry, bool is_append_log) {
+repl_req_ptr_t RaftStateMachine::localize_journal_entry_finish(nuraft::log_entry& lentry) {
     // Try to locate the rreq based on the log_entry.
     // If we are able to locate that req in the map for this entry, it could be one of
     //  a) This is an inline data and don't need any localization
@@ -165,11 +165,6 @@ repl_req_ptr_t RaftStateMachine::localize_journal_entry_finish(nuraft::log_entry
 
     auto rreq = m_rd.repl_key_to_req(rkey);
 
-    if (is_append_log) {
-        // make sure the rreq exists before appending the log, both leader and follower
-        RELEASE_ASSERT(rreq, "Failed to find in-memory rreq for repl_key={} before appending log", rkey.to_string());
-    }
-
     if ((rreq == nullptr) || (rreq->is_localize_pending())) {
         rreq = localize_journal_entry_prepare(lentry,
                                               -1 /* lsn=-1, since this is a finish call and we don't have lsn yet */);

src/lib/replication/repl_dev/raft_state_machine.h

@@ -124,7 +124,7 @@ class RaftStateMachine : public nuraft::state_machine {
     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, bool is_append_log = false);
+    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);