From 503deae0a73c307170f913961974eedbae496442 Mon Sep 17 00:00:00 2001 From: Patrick Ribbsaeter Date: Fri, 9 Oct 2026 03:52:43 +0200 Subject: [PATCH] Fix concurrent snapshot finalization race (#663) --- include/libnuraft/debugging_options.hxx | 9 ++ include/libnuraft/raft_server.hxx | 8 ++ src/handle_snapshot_sync.cxx | 38 +++++-- src/raft_server.cxx | 1 + tests/unit/snapshot_test.cxx | 140 ++++++++++++++++++++++++ 5 files changed, 187 insertions(+), 9 deletions(-) diff --git a/include/libnuraft/debugging_options.hxx b/include/libnuraft/debugging_options.hxx index 7471248..9e22efc 100644 --- a/include/libnuraft/debugging_options.hxx +++ b/include/libnuraft/debugging_options.hxx @@ -10,6 +10,7 @@ struct debugging_options { debugging_options() : disable_reconn_backoff_(false) , handle_cli_req_sleep_us_(0) + , snapshot_sync_req_sleep_us_(0) {} static debugging_options& get_instance() { @@ -29,6 +30,14 @@ struct debugging_options { * inside `handle_cli_req` function. */ std::atomic handle_cli_req_sleep_us_; + + /** + * If non-zero, the thread will sleep the given amount of time (in + * microseconds) after releasing the Raft lock to finalize a snapshot + * synchronization request. + * Used to reproduce overlapping finalization in deterministic tests. + */ + std::atomic snapshot_sync_req_sleep_us_; }; } diff --git a/include/libnuraft/raft_server.hxx b/include/libnuraft/raft_server.hxx index 8dd798a..70fb151 100644 --- a/include/libnuraft/raft_server.hxx +++ b/include/libnuraft/raft_server.hxx @@ -1558,6 +1558,14 @@ protected: */ std::atomic et_cnt_receiving_snapshot_; + /** + * Prevents more than one snapshot finalizer from crossing the unlocked + * state-machine pause boundary. + * + * Protected by `lock_`. + */ + bool snapshot_finalization_in_progress_; + /** * (Read-only) * The first snapshot distance. diff --git a/src/handle_snapshot_sync.cxx b/src/handle_snapshot_sync.cxx index 32b4a21..bafaa10 100644 --- a/src/handle_snapshot_sync.cxx +++ b/src/handle_snapshot_sync.cxx @@ -22,6 +22,7 @@ limitations under the License. #include "cluster_config.hxx" #include "context.hxx" +#include "debugging_options.hxx" #include "error_code.hxx" #include "event_awaiter.hxx" #include "exit_handler.hxx" @@ -33,7 +34,9 @@ limitations under the License. #include "tracer.hxx" #include +#include #include +#include namespace nuraft { @@ -465,6 +468,14 @@ bool raft_server::handle_snapshot_sync_req(snapshot_sync_req& req, std::unique_l try { // if offset == 0, it is the first object. bool is_first_obj = (req.get_offset()) ? false : true; + if (snapshot_finalization_in_progress_) { + p_wn( "decline snapshot request while finalization is in progress (idx %" PRIu64 + " term %" PRIu64 ", offset 0x%" PRIx64 ")", + req.get_snapshot().get_last_log_idx(), + req.get_snapshot().get_last_log_term(), + req.get_offset() ); + return false; + } bool is_last_obj = req.is_done(); if (is_first_obj || is_last_obj) { // INFO level: log only first and last object. @@ -527,9 +538,17 @@ bool raft_server::handle_snapshot_sync_req(snapshot_sync_req& req, std::unique_l } if (is_last_obj) { + snapshot_finalization_in_progress_ = true; + // let's pause committing in backgroud so it doesn't access logs // while they are being compacted guard.unlock(); + size_t test_sleep_us = debugging_options::get_instance() + .snapshot_sync_req_sleep_us_; + if (test_sleep_us) { + std::this_thread::sleep_for + ( std::chrono::microseconds(test_sleep_us) ); + } pause_state_machine_execution(); size_t wait_count = 0; while (!wait_for_state_machine_pause(500)) { @@ -537,6 +556,7 @@ bool raft_server::handle_snapshot_sync_req(snapshot_sync_req& req, std::unique_l ++wait_count); } guard.lock(); + snapshot_finalization_in_progress_ = false; struct ExecAutoResume { explicit ExecAutoResume(std::function func) : clean_func_(func) {} @@ -544,19 +564,19 @@ bool raft_server::handle_snapshot_sync_req(snapshot_sync_req& req, std::unique_l std::function clean_func_; } exec_auto_resume([this](){ resume_state_machine_execution(); }); + if (role_ != srv_role::follower) { + state_->set_receiving_snapshot(false); + ctx_->state_mgr_->save_state(*state_); + p_wn( "snapshot finalization became stale while waiting for the " + "state machine to pause, role %d", + static_cast(role_.load()) ); + return false; + } + state_->set_receiving_snapshot(false); ctx_->state_mgr_->save_state(*state_); p_in("clear receiving snapshot flag"); - // Only follower will run this piece of code, but let's check it again - if (role_ != srv_role::follower) { - // LCOV_EXCL_START - p_er("bad server role for applying a snapshot, exit for debugging"); - ctx_->state_mgr_->system_exit(raft_err::N11_not_follower_for_snapshot); - _sys_exit(-1); - // LCOV_EXCL_STOP - } - p_in( "successfully receive a snapshot (idx %" PRIu64 " term %" PRIu64 ") from leader", req.get_snapshot().get_last_log_idx(), diff --git a/src/raft_server.cxx b/src/raft_server.cxx index 01bd7a0..5e677fd 100644 --- a/src/raft_server.cxx +++ b/src/raft_server.cxx @@ -90,6 +90,7 @@ raft_server::raft_server(context* ctx, const init_options& opt) , log_store_(ctx->state_mgr_->load_log_store()) , state_machine_(ctx->state_machine_) , et_cnt_receiving_snapshot_(0) + , snapshot_finalization_in_progress_(false) , first_snapshot_distance_(0) , l_(ctx->logger_) , stale_config_(nullptr) diff --git a/tests/unit/snapshot_test.cxx b/tests/unit/snapshot_test.cxx index 6ce9b23..021b040 100644 --- a/tests/unit/snapshot_test.cxx +++ b/tests/unit/snapshot_test.cxx @@ -26,6 +26,9 @@ limitations under the License. #include +#include +#include + using namespace nuraft; using namespace raft_functional_common; @@ -33,6 +36,137 @@ using raft_result = cmd_result< ptr >; namespace snapshot_test { +static ptr make_final_snapshot_req(RaftPkg& src, + RaftPkg& dst, + ulong snapshot_idx) { + ptr snp = cs_new + ( snapshot_idx, + src.raftServer->get_term(), + dst.raftServer->get_config() ); + ptr data = buffer::alloc(1); + data->put((byte)0); + data->pos(0); + ptr sync_req = cs_new + ( snp, 0, data, true ); + ptr entry = cs_new + ( src.raftServer->get_term(), + sync_req->serialize(), + log_val_type::snp_sync_req ); + ptr req = cs_new + ( src.raftServer->get_term(), + msg_type::install_snapshot_request, + src.myId, + dst.myId, + src.raftServer->get_last_log_term(), + src.raftServer->get_last_log_idx(), + src.raftServer->get_committed_log_idx() ); + req->log_entries().push_back(entry); + return req; +} + +int snapshot_overlapping_finalization_test() { + reset_log_files(); + ptr f_base = cs_new(); + + RaftPkg s1(f_base, 1, "S1"); + RaftPkg s2(f_base, 2, "S2"); + RaftPkg s3(f_base, 3, "S3"); + std::vector pkgs = {&s1, &s2, &s3}; + + CHK_Z( launch_servers(pkgs) ); + CHK_Z( make_group(pkgs) ); + + ulong snapshot_idx = s2.raftServer->get_last_log_idx() + 1; + ptr first_req = make_final_snapshot_req(s1, s2, snapshot_idx); + ptr second_req = make_final_snapshot_req(s1, s2, snapshot_idx); + ptr first_resp; + + debugging_options::get_instance().snapshot_sync_req_sleep_us_ = 500 * 1000; + std::thread first_finalizer([&]() { + first_resp = s2.fNet->gotMsg(first_req); + }); + + size_t wait_count = 0; + while ( !s2.raftServer->is_receiving_snapshot() && + wait_count++ < 100 ) { + TestSuite::sleep_ms(5); + } + bool first_finalizer_active = s2.raftServer->is_receiving_snapshot(); + + std::chrono::steady_clock::time_point begin = + std::chrono::steady_clock::now(); + ptr second_resp = s2.fNet->gotMsg(second_req); + size_t elapsed_ms = std::chrono::duration_cast + ( std::chrono::steady_clock::now() - begin ).count(); + + CHK_NONNULL( second_resp.get() ); + CHK_FALSE( second_resp->get_accepted() ); + CHK_SM( elapsed_ms, (size_t)250 ); + + first_finalizer.join(); + debugging_options::get_instance().snapshot_sync_req_sleep_us_ = 0; + + CHK_TRUE( first_finalizer_active ); + CHK_NONNULL( first_resp.get() ); + + s1.raftServer->shutdown(); + s2.raftServer->shutdown(); + s3.raftServer->shutdown(); + f_base->destroy(); + + return 0; +} + +int snapshot_stale_finalization_test() { + reset_log_files(); + ptr f_base = cs_new(); + + RaftPkg s1(f_base, 1, "S1"); + RaftPkg s2(f_base, 2, "S2"); + RaftPkg s3(f_base, 3, "S3"); + std::vector pkgs = {&s1, &s2, &s3}; + + CHK_Z( launch_servers(pkgs) ); + CHK_Z( make_group(pkgs) ); + + ulong snapshot_idx = s2.raftServer->get_last_log_idx() + 1; + ptr req = make_final_snapshot_req(s1, s2, snapshot_idx); + ptr resp; + + debugging_options::get_instance().snapshot_sync_req_sleep_us_ = 1000 * 1000; + std::thread finalizer([&]() { + resp = s2.fNet->gotMsg(req); + }); + + size_t wait_count = 0; + while ( !s2.raftServer->is_receiving_snapshot() && + wait_count++ < 100 ) { + TestSuite::sleep_ms(5); + } + + bool takeover_requested = s2.raftServer->request_leadership(); + s2.fNet->execReqResp(); + s1.fTimer->invoke(timer_task_type::heartbeat_timer); + s1.fNet->execReqResp(); + s1.fNet->execReqResp(); + s2.fNet->execReqResp(); + + finalizer.join(); + debugging_options::get_instance().snapshot_sync_req_sleep_us_ = 0; + + CHK_TRUE( takeover_requested ); + CHK_TRUE( s2.raftServer->is_leader() ); + CHK_NONNULL( resp.get() ); + CHK_FALSE( resp->get_accepted() ); + + s1.raftServer->shutdown(); + s2.raftServer->shutdown(); + s3.raftServer->shutdown(); + f_base->destroy(); + + return 0; +} + int snapshot_basic_test() { reset_log_files(); ptr f_base = cs_new(); @@ -858,6 +992,12 @@ int main(int argc, char* argv[]) { ts.doTest( "snapshot leader switch test", snapshot_leader_switch_test ); + ts.doTest( "snapshot overlapping finalization test", + snapshot_overlapping_finalization_test ); + + ts.doTest( "snapshot stale finalization test", + snapshot_stale_finalization_test ); + #ifdef ENABLE_RAFT_STATS _msg("raft stats: ENABLED\n"); #else -- 2.55.0.windows.3