diff --git a/conanfile.py b/conanfile.py index d54bc7997..a183dee40 100644 --- a/conanfile.py +++ b/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" diff --git a/src/include/homestore/replication/repl_dev.h b/src/include/homestore/replication/repl_dev.h index 31c51e8d1..d7166ff9a 100644 --- a/src/include/homestore/replication/repl_dev.h +++ b/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); } }; diff --git a/src/lib/replication/log_store/repl_log_store.cpp b/src/lib/replication/log_store/repl_log_store.cpp index de933da3e..d3085da55 100644 --- a/src/lib/replication/log_store/repl_log_store.cpp +++ b/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); diff --git a/src/lib/replication/repl_dev/raft_repl_dev.cpp b/src/lib/replication/repl_dev/raft_repl_dev.cpp index 07ebd98cd..c275b5c7b 100644 --- a/src/lib/replication/repl_dev/raft_repl_dev.cpp +++ b/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, ""}; diff --git a/src/lib/replication/repl_dev/raft_state_machine.cpp b/src/lib/replication/repl_dev/raft_state_machine.cpp index 94b7ff275..3db01569a 100644 --- a/src/lib/replication/repl_dev/raft_state_machine.cpp +++ b/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 */); diff --git a/src/lib/replication/repl_dev/raft_state_machine.h b/src/lib/replication/repl_dev/raft_state_machine.h index d7057ed56..e08756155 100644 --- a/src/lib/replication/repl_dev/raft_state_machine.h +++ b/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);