Skip to content

Fix data races on Aft leadership state accessors #8249

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
Amaury Chamayou (achamayou) wants to merge 3 commits into
base: main
Choose a base branch
from
+104 −32
Open

Fix data races on Aft leadership state accessors #8249

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
a036456
Fix data race on Aft leadership_state
achamayou Aug 31, 2026
c96d5dc
Guard Aft::primary() with state->lock
achamayou Sep 1, 2026
15bb471
Merge branch 'main' into achamayou-atomic-leadership-state
achamayou Sep 2, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
59 changes: 58 additions & 1 deletion src/consensus/aft/impl/state.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -9,6 +9,7 @@
#include "ds/internal_logger.h"
#include "kv/kv_types.h"

#include <atomic>
#include <deque>
#include <map>
#include <set>
Expand Down Expand Up @@ -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<ccf::kv::LeadershipState> 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<ccf::kv::LeadershipState>());
}

struct State
{
State(ccf::NodeId node_id_, bool pre_vote_enabled_ = true) :
Expand Down Expand Up @@ -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;

Expand Down
77 changes: 46 additions & 31 deletions src/consensus/aft/raft.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -249,6 +249,11 @@ namespace aft

std::optional<ccf::NodeId> 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<ccf::pal::Mutex> guard(state->lock);
return leader_id;
}

Expand All @@ -259,12 +264,13 @@ namespace aft

bool is_primary() override
{
return state->leadership_state == ccf::kv::LeadershipState::Leader;
return state->leadership_state.load() == ccf::kv::LeadershipState::Leader;
}
Comment thread
achamayou marked this conversation as resolved.

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
Expand All @@ -285,7 +291,8 @@ namespace aft
return false;
}
std::unique_lock<ccf::pal::Mutex> 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);
}

Expand All @@ -305,7 +312,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
Expand Down Expand Up @@ -602,7 +610,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())
{
Expand All @@ -625,7 +633,7 @@ namespace aft
{
std::lock_guard<ccf::pal::Mutex> 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());
Expand Down Expand Up @@ -691,7 +699,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)
Expand Down Expand Up @@ -837,7 +845,7 @@ namespace aft
std::unique_lock<ccf::pal::Mutex> 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)
{
Expand Down Expand Up @@ -981,13 +989,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();
}

Expand Down Expand Up @@ -1138,8 +1148,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);
}
Expand Down Expand Up @@ -1607,7 +1619,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",
Expand Down Expand Up @@ -1988,8 +2000,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",
Expand All @@ -2000,7 +2013,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)
Expand All @@ -2014,7 +2027,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.
Expand Down Expand Up @@ -2117,7 +2131,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();
Expand Down Expand Up @@ -2162,7 +2176,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;
Expand Down Expand Up @@ -2219,7 +2233,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;

Expand Down Expand Up @@ -2273,7 +2287,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,
Expand Down Expand Up @@ -2363,7 +2377,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);
Expand Down Expand Up @@ -2396,7 +2410,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;
Expand Down Expand Up @@ -2448,7 +2462,7 @@ namespace aft

if (is_elected)
{
switch (state->leadership_state)
switch (state->leadership_state.load())
{
case ccf::kv::LeadershipState::PreVoteCandidate:
become_candidate();
Expand All @@ -2471,7 +2485,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");
Expand Down Expand Up @@ -2782,7 +2796,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",
Expand Down Expand Up @@ -2850,7 +2864,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);
}
Expand Down