1. fix handle_error and only trigger handle_error for timeout rreqs in handle_raft_event

2. include concept 'volatile' vs 'non-volatile' for log
3. update replay logic : add BLK_ALLOCATED and DATA_RECEIVED only when data linked and received

Author: yawzhang

conanfile.py

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

src/lib/replication/log_store/repl_log_store.cpp

@@ -17,7 +17,7 @@ uint64_t ReplLogStore::append(nuraft::ptr< nuraft::log_entry >& entry) {
 
     repl_req_ptr_t rreq = m_sm.localize_journal_entry_finish(*entry);
     ulong lsn = HomeRaftLogStore::append(entry);
-    m_sm.link_lsn_to_req(rreq, int64_cast(lsn));
+    m_sm.link_lsn_to_req(rreq, m_sm.to_volatile_lsn(lsn));
 
     RD_LOGD("Raft Channel: Received append log entry rreq=[{}]", rreq->to_compact_string());
     return lsn;
@@ -32,7 +32,7 @@ void ReplLogStore::write_at(ulong index, nuraft::ptr< nuraft::log_entry >& entry
 
     repl_req_ptr_t rreq = m_sm.localize_journal_entry_finish(*entry);
     HomeRaftLogStore::write_at(index, entry);
-    m_sm.link_lsn_to_req(rreq, int64_cast(index));
+    m_sm.link_lsn_to_req(rreq, m_sm.to_volatile_lsn(index));
     RD_LOGD("Raft Channel: Received write_at log entry rreq=[{}]", rreq->to_compact_string());
 }
 
@@ -43,7 +43,7 @@ void ReplLogStore::end_of_append_batch(ulong start_lsn, ulong count) {
     auto reqs = sisl::VectorPool< repl_req_ptr_t >::alloc();
     auto proposer_reqs = sisl::VectorPool< repl_req_ptr_t >::alloc();
     for (int64_t lsn = int64_cast(start_lsn); lsn <= end_lsn; ++lsn) {
-        auto rreq = m_sm.lsn_to_req(lsn);
+        auto rreq = m_sm.lsn_to_req(m_sm.to_volatile_lsn(lsn));
         // Skip this call in proposer, since this method will synchronously flush the data, which is not required for
         // leader. Proposer will call the flush as part of commit after receiving quorum, upon which time, there is a
         // high possibility the log entry is already flushed. Skip it for rreq == nullptr which is the case for raft
@@ -93,6 +93,17 @@ void ReplLogStore::end_of_append_batch(ulong start_lsn, ulong count) {
             if (rreq) { rreq->add_state(repl_req_state_t::LOG_FLUSHED); }
         }
     }
+
+    // Convert volatile logs to non-volatile logs in state machine
+    for (int64_t lsn = int64_cast(start_lsn); lsn <= end_lsn; ++lsn) {
+        auto volatile_lsn = m_sm.to_volatile_lsn(lsn);
+        auto rreq = m_sm.lsn_to_req(volatile_lsn);
+        if (rreq != nullptr) {
+            m_sm.link_lsn_to_req(rreq, lsn);
+            m_sm.unlink_lsn_to_req(volatile_lsn, rreq);
+        }
+    }
+
     sisl::VectorPool< repl_req_ptr_t >::free(reqs);
     sisl::VectorPool< repl_req_ptr_t >::free(proposer_reqs);
 }

src/lib/replication/repl_dev/common.cpp

@@ -105,7 +105,7 @@ raft_buf_ptr_t& repl_req_ctx::raft_journal_buf() { return std::get< raft_buf_ptr
 uint8_t* repl_req_ctx::raw_journal_buf() { return std::get< std::unique_ptr< uint8_t[] > >(m_journal_buf).get(); }
 
 void repl_req_ctx::set_lsn(int64_t lsn) {
-    DEBUG_ASSERT((m_lsn == -1) || (m_lsn == lsn),
+    DEBUG_ASSERT((m_lsn == -1) || (m_lsn == lsn) || (m_lsn == -lsn),
                  "Changing lsn for request={} on the fly can cause race condition, not expected. lsn {}, m_lsn {}",
                  to_string(), lsn, m_lsn);
     m_lsn = lsn;

src/lib/replication/repl_dev/raft_repl_dev.cpp

@@ -594,7 +594,8 @@ folly::Future< folly::Unit > RaftReplDev::notify_after_data_written(std::vector<
     });
 }
 
-bool RaftReplDev::wait_for_data_receive(std::vector< repl_req_ptr_t > const& rreqs, uint64_t timeout_ms) {
+bool RaftReplDev::wait_for_data_receive(std::vector< repl_req_ptr_t > const& rreqs, uint64_t timeout_ms,
+                                        std::vector< repl_req_ptr_t >* timeout_rreqs) {
     std::vector< folly::Future< folly::Unit > > futs;
     std::vector< repl_req_ptr_t > only_wait_reqs;
     only_wait_reqs.reserve(rreqs.size());
@@ -621,14 +622,23 @@ bool RaftReplDev::wait_for_data_receive(std::vector< repl_req_ptr_t > const& rre
 
     // We are yet to support reactive fetch from remote.
     if (is_resync_mode()) {
-        check_and_fetch_remote_data(std::move(only_wait_reqs));
+        check_and_fetch_remote_data(only_wait_reqs);
     } else {
-        m_repl_svc.add_to_fetch_queue(shared_from_this(), std::move(only_wait_reqs));
+        m_repl_svc.add_to_fetch_queue(shared_from_this(), only_wait_reqs);
     }
 
     // block waiting here until all the futs are ready (data channel filled in and promises are made);
-    auto all_futs = folly::collectAllUnsafe(futs).wait(std::chrono::milliseconds(timeout_ms));
-    return (all_futs.isReady());
+    auto all_futs_ready = folly::collectAllUnsafe(futs).wait(std::chrono::milliseconds(timeout_ms)).isReady();
+    if (!all_futs_ready && timeout_rreqs != nullptr) {
+        timeout_rreqs->clear();
+        for (size_t i{0}; i < futs.size(); ++i) {
+            if (!futs[i].isReady()) {
+                timeout_rreqs->emplace_back(only_wait_reqs[i]);
+            }
+        }
+        all_futs_ready = timeout_rreqs->empty();
+    }
+    return all_futs_ready;
 }
 
 void RaftReplDev::check_and_fetch_remote_data(std::vector< repl_req_ptr_t > rreqs) {
@@ -953,18 +963,22 @@ void RaftReplDev::handle_commit(repl_req_ptr_t rreq, bool recovery) {
 
 void RaftReplDev::handle_error(repl_req_ptr_t const& rreq, ReplServiceError err) {
     if (err == ReplServiceError::OK) { return; }
+    RD_LOGE("Raft Channel: Error in processing rreq=[{}] error={}", rreq->to_string(), err);
 
     if (!rreq->add_state_if_not_already(repl_req_state_t::ERRORED)) {
-        RD_LOGE("Raft Channel: Error in processing rreq=[{}] error={}", rreq->to_string(), err);
+        RD_LOGE("Raft Channel: Error has been added for rreq=[{}] error={}", rreq->to_string(), err);
         return;
     }
 
     // Remove from the map and thus its no longer accessible from applier_create_req
     m_repl_key_req_map.erase(rreq->rkey());
 
-    if (rreq->op_code() == journal_type_t::HS_DATA_INLINED) {
+    // Ensure non-volatile lsn not exist because handle_error should not be called after append entries.
+    HS_REL_ASSERT(m_state_machine->lsn_to_req(rreq->lsn()) == nullptr,
+                  "Unexpected: LSN={} is already ready to commit", rreq->lsn());
+
+    if (rreq->op_code() == journal_type_t::HS_DATA_LINKED) {
         // Free the blks which is allocated already
-        RD_LOGE("Raft Channel: Error in processing rreq=[{}] error={}", rreq->to_string(), err);
         if (rreq->has_state(repl_req_state_t::BLK_ALLOCATED)) {
             auto blkid = rreq->local_blkid();
             data_service().async_free_blk(blkid).thenValue([blkid](auto&& err) {
@@ -1276,8 +1290,9 @@ std::pair< bool, nuraft::cb_func::ReturnCode > RaftReplDev::handle_raft_event(nu
             }
 
             // Wait till we receive the data from its originator for all the requests
-            if (!wait_for_data_receive(*reqs, HS_DYNAMIC_CONFIG(consensus.data_receive_timeout_ms))) {
-                for (auto const& rreq : *reqs) {
+            std::vector< repl_req_ptr_t > timeout_rreqs;
+            if (!wait_for_data_receive(*reqs, HS_DYNAMIC_CONFIG(consensus.data_receive_timeout_ms), &timeout_rreqs)) {
+                for (auto const& rreq : timeout_rreqs) {
                     handle_error(rreq, ReplServiceError::TIMEOUT);
                 }
                 ret = nuraft::cb_func::ReturnCode::ReturnNull;
@@ -1468,20 +1483,22 @@ void RaftReplDev::on_log_found(logstore_seq_num_t lsn, log_buffer buf, void* ctx
     RD_DBG_ASSERT(happened, "rreq already exists for rkey={}", rkey.to_string());
     uint32_t data_size{0u};
 
+    // If the data is linked and value_size is non-zero, it means blks have been allocated for data.
+    // Since the log is flushed after data is written, the data has already been received.
     if ((jentry->code == journal_type_t::HS_DATA_LINKED) && (jentry->value_size > 0)) {
         MultiBlkId entry_blkid;
         entry_blkid.deserialize(entry_to_val(jentry), true /* copy */);
         data_size = entry_blkid.blk_count() * get_blk_size();
         rreq->set_local_blkid(entry_blkid);
+        rreq->add_state(repl_req_state_t::BLK_ALLOCATED);
+        rreq->add_state(repl_req_state_t::DATA_RECEIVED);
     }
 
     rreq->set_lsn(repl_lsn);
     // keep lentry in scope for the lyfe cycle of the rreq
     rreq->set_lentry(lentry);
     rreq->init(rkey, jentry->code, false /* is_proposer */, entry_to_hdr(jentry), entry_to_key(jentry), data_size);
     // we load the log from log device, implies log flushed.  We only flush log after data is written to data device.
-    rreq->add_state(repl_req_state_t::BLK_ALLOCATED);
-    rreq->add_state(repl_req_state_t::DATA_RECEIVED);
     rreq->add_state(repl_req_state_t::DATA_WRITTEN);
     rreq->add_state(repl_req_state_t::LOG_RECEIVED);
     rreq->add_state(repl_req_state_t::LOG_FLUSHED);

src/lib/replication/repl_dev/raft_repl_dev.h

@@ -279,8 +279,15 @@ class RaftReplDev : public ReplDev,
     void fetch_data_from_remote(std::vector< repl_req_ptr_t > rreqs);
     void handle_fetch_data_response(sisl::GenericClientResponse response, std::vector< repl_req_ptr_t > rreqs);
     bool is_resync_mode();
+
+    /**
+     * \brief This method handles errors that occur during append entries or data receiving.
+     * It should not be called after the append entries phase.
+     */
     void handle_error(repl_req_ptr_t const& rreq, ReplServiceError err);
-    bool wait_for_data_receive(std::vector< repl_req_ptr_t > const& rreqs, uint64_t timeout_ms);
+
+    bool wait_for_data_receive(std::vector < repl_req_ptr_t > const &rreqs, uint64_t timeout_ms,
+                               std::vector < repl_req_ptr_t > *timeout_rreqs = nullptr);
     void on_log_found(logstore_seq_num_t lsn, log_buffer buf, void* ctx);
     void commit_blk(repl_req_ptr_t rreq);
     void replace_member(repl_req_ptr_t rreq);

src/lib/replication/repl_dev/raft_state_machine.cpp

@@ -179,7 +179,7 @@ repl_req_ptr_t RaftStateMachine::localize_journal_entry_finish(nuraft::log_entry
 raft_buf_ptr_t RaftStateMachine::pre_commit_ext(nuraft::state_machine::ext_op_params const& params) {
     int64_t lsn = s_cast< int64_t >(params.log_idx);
 
-    repl_req_ptr_t rreq = lsn_to_req(lsn);
+    repl_req_ptr_t rreq = lsn_to_req(to_volatile_lsn(lsn));
     RD_LOGD("Raft channel: Precommit rreq=[{}]", rreq->to_compact_string());
     m_rd.m_listener->on_pre_commit(rreq->lsn(), rreq->header(), rreq->key(), rreq);
 
@@ -293,8 +293,11 @@ void RaftStateMachine::link_lsn_to_req(repl_req_ptr_t rreq, int64_t lsn) {
     rreq->add_state(repl_req_state_t::LOG_RECEIVED);
     // reset the rreq created_at time to now https://github.com/eBay/HomeStore/issues/506
     rreq->set_created_time();
-    [[maybe_unused]] auto r = m_lsn_req_map.insert(lsn, std::move(rreq));
-    RD_DBG_ASSERT_EQ(r.second, true, "lsn={} already in precommit list, exist_term={}", lsn, r.first->second->term());
+    auto r = m_lsn_req_map.insert(lsn, std::move(rreq));
+    if (!r.second) {
+        RD_LOG(ERROR, "lsn={} already in precommit list, exist_term={}", lsn, r.first->second->term());
+        // TODO: we need to think about the case where volatile is in the map already, is it safe to overwrite it?
+    }
 }
 
 repl_req_ptr_t RaftStateMachine::lsn_to_req(int64_t lsn) {
@@ -400,4 +403,9 @@ void RaftStateMachine::free_user_snp_ctx(void*& user_snp_ctx) { m_rd.m_listener-
 
 std::string RaftStateMachine::rdev_name() const { return m_rd.rdev_name(); }
 
+int64_t RaftStateMachine::to_volatile_lsn(ulong log_idx) {
+    RELEASE_ASSERT(log_idx <= INT64_MAX, "lsn={} is greater than INT64_MAX", log_idx);
+    return - int64_cast(log_idx);
+}
+
 } // namespace homestore

src/lib/replication/repl_dev/raft_state_machine.h

@@ -145,6 +145,14 @@ class RaftStateMachine : public nuraft::state_machine {
 
     static bool is_hs_snp_obj(uint64_t obj_id) { return (obj_id & snp_obj_id_type_app) == 0; }
 
+    // There are two types of logs in the m_lsn_req_map: volatile and non-volatile.
+    // A log is considered volatile until it is flushed to the log store at the end of the append operation.
+    // Before this flushing occurs, the log and its associated request can be treated as volatile.
+    // Therefore, the pre_commit operation should utilize the volatile log.
+    // Once the log is flushed to the log store, it converts to a non-volatile state.
+    // At this point, the commit operation can use the non-volatile one.
+    int64_t to_volatile_lsn(ulong log_idx);
+
 private:
     void after_precommit_in_leader(const nuraft::raft_server::req_ext_cb_params& params);
 };