From a036456c8868af317e95cd69f0747a5f5d9f271b Mon Sep 17 00:00:00 2001 From: achamayou Date: Mon, 31 Aug 2026 19:23:52 +0100 Subject: [PATCH 1/2] Fix data race on Aft leadership_state Aft::is_primary(), is_candidate() and their callers read state->leadership_state without holding state->lock, while every write to it is made under that lock during election transitions. ThreadSanitizer reports this as a data race against a concurrent election. Taking state->lock in those accessors is not an option. Store::commit() calls is_primary() while holding the KV version lock, and Aft calls into the Store from under state->lock (become_leader, compact, rollback), so locking there would invert an existing lock order. Make the value atomic instead, so the unsynchronised reads are well defined for all seven call sites rather than just the one TSAN happened to hit. std::atomic is neither copyable nor movable, and so cannot be a field of a type declared with the DECLARE_JSON_* macros, which round-trip each field by value; a small wrapper restores value semantics for serialisation while keeping every access atomic. The serialised form is unchanged. No behaviour change. Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> Copilot-Session: 75d99c5d-6efa-4048-8032-8c78b97208d9 --- src/consensus/aft/impl/state.h | 59 +++++++++++++++++++++++++++- src/consensus/aft/raft.h | 72 +++++++++++++++++++--------------- 2 files changed, 99 insertions(+), 32 deletions(-) diff --git a/src/consensus/aft/impl/state.h b/src/consensus/aft/impl/state.h index 248cb34ab149..a11117257d39 100644 --- a/src/consensus/aft/impl/state.h +++ b/src/consensus/aft/impl/state.h @@ -9,6 +9,7 @@ #include "ds/internal_logger.h" #include "kv/kv_types.h" +#include #include #include #include @@ -130,6 +131,54 @@ namespace aft } }; + // std::atomic is neither copyable nor movable, so it cannot be a field of a + // type declared with the DECLARE_JSON_* macros, which round-trip each field + // by value. This wrapper restores value semantics for serialisation while + // keeping every access to the underlying value atomic. + class AtomicLeadershipState + { + std::atomic value; + + public: + AtomicLeadershipState( + ccf::kv::LeadershipState value_ = ccf::kv::LeadershipState::None) : + value(value_) + {} + + AtomicLeadershipState(const AtomicLeadershipState& other) : + value(other.load()) + {} + + AtomicLeadershipState& operator=(const AtomicLeadershipState& other) + { + if (this != &other) + { + store(other.load()); + } + return *this; + } + + [[nodiscard]] ccf::kv::LeadershipState load() const + { + return value.load(std::memory_order_acquire); + } + + void store(ccf::kv::LeadershipState value_) + { + value.store(value_, std::memory_order_release); + } + }; + + inline void to_json(nlohmann::json& j, const AtomicLeadershipState& state) + { + j = state.load(); + } + + inline void from_json(const nlohmann::json& j, AtomicLeadershipState& state) + { + state.store(j.get()); + } + struct State { State(ccf::NodeId node_id_, bool pre_vote_enabled_ = true) : @@ -161,7 +210,15 @@ namespace aft // the node // Leader -> Follower, when receiving entries for a newer term // Candidate -> Follower, when receiving entries for a newer term - ccf::kv::LeadershipState leadership_state = ccf::kv::LeadershipState::None; + // + // Written only under `lock`, but read without it by the unsynchronised + // accessors (is_primary(), is_candidate(), ...) which callers outside + // consensus rely on. Atomic so those reads are not data races. Taking + // `lock` in those accessors instead is not an option: Store::commit() + // calls is_primary() while holding the KV version lock, and Aft calls + // into the Store from under `lock`, so locking here would invert an + // existing lock order. + AtomicLeadershipState leadership_state; ccf::kv::MembershipState membership_state = ccf::kv::MembershipState::Active; diff --git a/src/consensus/aft/raft.h b/src/consensus/aft/raft.h index 9056c818e5ff..2fc612d663c6 100644 --- a/src/consensus/aft/raft.h +++ b/src/consensus/aft/raft.h @@ -31,25 +31,25 @@ CCF_LOG_FMT(TRACE, "raft") \ ("{} | {} | {} | " s, \ state->node_id, \ - state->leadership_state, \ + state->leadership_state.load(), \ state->membership_state __VA_OPT__(, ) __VA_ARGS__) # define RAFT_DEBUG_FMT(s, ...) \ CCF_LOG_FMT(DEBUG, "raft") \ ("{} | {} | {} | " s, \ state->node_id, \ - state->leadership_state, \ + state->leadership_state.load(), \ state->membership_state __VA_OPT__(, ) __VA_ARGS__) # define RAFT_INFO_FMT(s, ...) \ CCF_LOG_FMT(INFO, "raft") \ ("{} | {} | {} | " s, \ state->node_id, \ - state->leadership_state, \ + state->leadership_state.load(), \ state->membership_state __VA_OPT__(, ) __VA_ARGS__) # define RAFT_FAIL_FMT(s, ...) \ CCF_LOG_FMT(FAIL, "raft") \ ("{} | {} | {} | " s, \ state->node_id, \ - state->leadership_state, \ + state->leadership_state.load(), \ state->membership_state __VA_OPT__(, ) __VA_ARGS__) #else # define RAFT_TRACE_FMT LOG_TRACE_FMT @@ -259,12 +259,13 @@ namespace aft bool is_primary() override { - return state->leadership_state == ccf::kv::LeadershipState::Leader; + return state->leadership_state.load() == ccf::kv::LeadershipState::Leader; } bool is_candidate() override { - return state->leadership_state == ccf::kv::LeadershipState::Candidate; + return state->leadership_state.load() == + ccf::kv::LeadershipState::Candidate; } bool can_replicate() override @@ -285,7 +286,8 @@ namespace aft return false; } std::unique_lock guard(state->lock); - return state->leadership_state == ccf::kv::LeadershipState::Leader && + return state->leadership_state.load() == + ccf::kv::LeadershipState::Leader && (state->last_idx - state->commit_idx >= max_uncommitted_tx_count); } @@ -305,7 +307,8 @@ namespace aft bool is_backup() override { - return state->leadership_state == ccf::kv::LeadershipState::Follower; + return state->leadership_state.load() == + ccf::kv::LeadershipState::Follower; } bool is_active() const @@ -602,7 +605,7 @@ namespace aft details.primary_id = leader_id; details.current_view = state->current_view; details.ticking = ticking; - details.leadership_state = state->leadership_state; + details.leadership_state = state->leadership_state.load(); details.membership_state = state->membership_state; if (is_retired()) { @@ -625,7 +628,7 @@ namespace aft { std::lock_guard guard(state->lock); - if (state->leadership_state != ccf::kv::LeadershipState::Leader) + if (state->leadership_state.load() != ccf::kv::LeadershipState::Leader) { RAFT_DEBUG_FMT( "Failed to replicate {} items: not leader", entries.size()); @@ -691,7 +694,7 @@ namespace aft RAFT_DEBUG_FMT( "membership: {} leadership: {}", state->membership_state, - state->leadership_state); + state->leadership_state.load()); if ( state->membership_state == ccf::kv::MembershipState::Retired && state->retirement_phase == ccf::kv::RetirementPhase::Ordered) @@ -837,7 +840,7 @@ namespace aft std::unique_lock guard(state->lock); timeout_elapsed += elapsed; - if (state->leadership_state == ccf::kv::LeadershipState::Leader) + if (state->leadership_state.load() == ccf::kv::LeadershipState::Leader) { if (timeout_elapsed >= request_timeout) { @@ -981,13 +984,15 @@ namespace aft bool can_replicate_unsafe() { - return state->leadership_state == ccf::kv::LeadershipState::Leader && + return state->leadership_state.load() == + ccf::kv::LeadershipState::Leader && !is_retired_committed(); } bool can_sign_unsafe() { - return state->leadership_state == ccf::kv::LeadershipState::Leader && + return state->leadership_state.load() == + ccf::kv::LeadershipState::Leader && !is_retired_committed(); } @@ -1138,8 +1143,10 @@ namespace aft // follower if necessary if ( state->current_view == r.term && - (state->leadership_state == ccf::kv::LeadershipState::Candidate || - state->leadership_state == ccf::kv::LeadershipState::PreVoteCandidate)) + (state->leadership_state.load() == + ccf::kv::LeadershipState::Candidate || + state->leadership_state.load() == + ccf::kv::LeadershipState::PreVoteCandidate)) { become_aware_of_new_term(r.term); } @@ -1607,7 +1614,7 @@ namespace aft #endif // Ignore if we're not the leader. - if (state->leadership_state != ccf::kv::LeadershipState::Leader) + if (state->leadership_state.load() != ccf::kv::LeadershipState::Leader) { RAFT_INFO_FMT( "Recv {} to {} from {}: no longer leader", @@ -1988,8 +1995,9 @@ namespace aft } if ( - state->leadership_state != ccf::kv::LeadershipState::PreVoteCandidate && - state->leadership_state != ccf::kv::LeadershipState::Candidate) + state->leadership_state.load() != + ccf::kv::LeadershipState::PreVoteCandidate && + state->leadership_state.load() != ccf::kv::LeadershipState::Candidate) { RAFT_INFO_FMT( "Recv {} to {} from: {}: we aren't a candidate", @@ -2000,7 +2008,7 @@ namespace aft } if ( election_type == ElectionType::RegularVote && - state->leadership_state != ccf::kv::LeadershipState::Candidate) + state->leadership_state.load() != ccf::kv::LeadershipState::Candidate) { // Stale message from previous candidacy // Candidate(T) -> Follower(T) -> PreVoteCandidate(T) @@ -2014,7 +2022,8 @@ namespace aft } if ( election_type == ElectionType::PreVote && - state->leadership_state != ccf::kv::LeadershipState::PreVoteCandidate) + state->leadership_state.load() != + ccf::kv::LeadershipState::PreVoteCandidate) { // To receive a PreVoteResponse, we must have been a PreVoteCandidate in // that term. @@ -2117,7 +2126,7 @@ namespace aft return; } - state->leadership_state = ccf::kv::LeadershipState::PreVoteCandidate; + state->leadership_state.store(ccf::kv::LeadershipState::PreVoteCandidate); leader_id.reset(); reset_votes_for_me(); @@ -2162,7 +2171,7 @@ namespace aft return; } - state->leadership_state = ccf::kv::LeadershipState::Candidate; + state->leadership_state.store(ccf::kv::LeadershipState::Candidate); leader_id.reset(); voted_for = state->node_id; @@ -2219,7 +2228,7 @@ namespace aft store->initialise_term(state->current_view); } - state->leadership_state = ccf::kv::LeadershipState::Leader; + state->leadership_state.store(ccf::kv::LeadershipState::Leader); leader_id = state->node_id; should_sign = true; @@ -2273,7 +2282,7 @@ namespace aft restart_election_timeout(); reset_last_ack_timeouts(); - state->leadership_state = ccf::kv::LeadershipState::Follower; + state->leadership_state.store(ccf::kv::LeadershipState::Follower); RAFT_INFO_FMT( "Becoming follower {}: {}.{}", state->node_id, @@ -2363,7 +2372,7 @@ namespace aft RAFT_INFO_FMT( "Becoming retired, phase {} (leadership {}): {}: {} at {}", phase, - state->leadership_state, + state->leadership_state.load(), state->node_id, state->current_view, idx); @@ -2396,7 +2405,7 @@ namespace aft nominate_successor(); leader_id.reset(); - state->leadership_state = ccf::kv::LeadershipState::None; + state->leadership_state.store(ccf::kv::LeadershipState::None); } state->membership_state = ccf::kv::MembershipState::Retired; @@ -2448,7 +2457,7 @@ namespace aft if (is_elected) { - switch (state->leadership_state) + switch (state->leadership_state.load()) { case ccf::kv::LeadershipState::PreVoteCandidate: become_candidate(); @@ -2471,7 +2480,7 @@ namespace aft // idx. void update_commit() { - if (state->leadership_state != ccf::kv::LeadershipState::Leader) + if (state->leadership_state.load() != ccf::kv::LeadershipState::Leader) { throw std::logic_error( "update_commit() must only be called while this node is leader"); @@ -2782,7 +2791,7 @@ namespace aft void nominate_successor() override { - if (state->leadership_state != ccf::kv::LeadershipState::Leader) + if (state->leadership_state.load() != ccf::kv::LeadershipState::Leader) { RAFT_DEBUG_FMT( "Not proposing request vote from {} since not leader", @@ -2850,7 +2859,8 @@ namespace aft all_other_nodes.try_emplace( node_info.first, node_info.second, index, 0); - if (state->leadership_state == ccf::kv::LeadershipState::Leader) + if ( + state->leadership_state.load() == ccf::kv::LeadershipState::Leader) { send_append_entries(node_info.first, index); } From c96d5dc620754001dc01ef23fc9339158c53e327 Mon Sep 17 00:00:00 2001 From: achamayou Date: Tue, 1 Sep 2026 07:30:06 +0100 Subject: [PATCH 2/2] Guard Aft::primary() with state->lock leader_id is written under state->lock on every leadership transition, and read under it by get_details(), but Aft::primary() returned it with no synchronisation. Same class of race as leadership_state, in the adjacent accessor. It cannot be made atomic, being a std::optional, but it does not need to be: unlike is_primary(), primary() is not called from Store::commit() under the KV version lock, and it is never called from inside Aft, so taking state->lock here introduces no ordering risk. The adjacent can_replicate() already takes the same lock from the same callers. Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> Copilot-Session: 75d99c5d-6efa-4048-8032-8c78b97208d9 --- src/consensus/aft/raft.h | 5 +++++ 1 file changed, 5 insertions(+) diff --git a/src/consensus/aft/raft.h b/src/consensus/aft/raft.h index 2fc612d663c6..1b9f639970b1 100644 --- a/src/consensus/aft/raft.h +++ b/src/consensus/aft/raft.h @@ -249,6 +249,11 @@ namespace aft std::optional primary() override { + // leader_id is written under state->lock, and unlike is_primary() this + // is not called from Store::commit() under the KV version lock, so it + // can be read under the lock rather than made atomic. std::optional< + // NodeId> could not be made atomic in any case. + std::lock_guard guard(state->lock); return leader_id; }