myxce/src/lib/raft/raft.c

270 lines
8.6 KiB
C

#include "raft.h"
void raft_snapshot(const raft_state_t *s, raft_snapshot_t *snap)
{
snap->term = s->term;
snap->role = s->role;
snap->voted_for = s->voted_for;
snap->voted_for_term = s->voted_for_term;
snap->votes_granted = s->votes_granted;
snap->election_decided = s->election_decided;
}
int raft_init(raft_state_t *s, const raft_config_t *cfg)
{
/* Validate required callbacks and n_peers before touching *s */
if (cfg->n_peers < 1)
return -EINVAL;
/* Even cluster sizes (>1) have no fault-tolerance advantage over the next
* smaller odd size and can mislead operators into thinking split-brain is
* impossible. Reject to prevent misconfiguration. */
if (cfg->n_peers > 1 && cfg->n_peers % 2 == 0)
return -EINVAL;
if (!cfg->cbs)
return -EINVAL;
if (!cfg->cbs->on_send_vote_request)
return -EINVAL;
if (!cfg->cbs->on_become_leader)
return -EINVAL;
if (!cfg->cbs->on_become_follower)
return -EINVAL;
if (!cfg->cbs->on_persist)
return -EINVAL;
if (!cfg->cbs->on_schedule_timeout)
return -EINVAL;
if (!cfg->cbs->on_cancel_timeout)
return -EINVAL;
if (!cfg->cbs->on_start_election_after_delay)
return -EINVAL;
s->term = cfg->initial_term;
s->role = RAFT_FOLLOWER;
s->voted_for = cfg->initial_voted_for;
s->voted_for_term = 0;
s->n_peers = cfg->n_peers;
s->self_idx = cfg->self_idx;
s->votes_granted = 0;
s->replies_received = 0;
s->election_decided = 0;
s->election_timeout_ms = cfg->election_timeout_ms ? cfg->election_timeout_ms : 2000;
s->my_priority = cfg->my_priority;
s->cbs = cfg->cbs;
s->ctx = cfg->ctx;
return 0;
}
int raft_start_election(raft_state_t *s)
{
/* Guard: reject if already CANDIDATE or LEADER */
if (s->role == RAFT_CANDIDATE || s->role == RAFT_LEADER)
return -EALREADY;
/* Increment term and record self-vote before any RPCs (split-brain prevention) */
s->term++;
s->voted_for = s->self_idx;
s->voted_for_term = s->term;
s->votes_granted = 1; /* self-vote */
s->replies_received = 0;
s->election_decided = 0;
if (s->cbs->on_trace)
s->cbs->on_trace(s->ctx, "term_update", NULL);
if (s->cbs->on_trace)
s->cbs->on_trace(s->ctx, "self_vote", NULL);
/* Persist new term and self-vote before sending any RPCs */
s->cbs->on_persist(s->ctx, s->term, s->self_idx);
if (s->cbs->on_trace)
s->cbs->on_trace(s->ctx, "election_start", NULL);
/* Single-node shortcircuit: win immediately, no RPCs, no timeout */
if (s->n_peers == 1) {
s->role = RAFT_LEADER;
s->election_decided = 1;
s->cbs->on_become_leader(s->ctx, s->term);
return RAFT_ELECTED_IMMEDIATELY;
}
/* Multi-node: transition to CANDIDATE */
s->role = RAFT_CANDIDATE;
/* Send RequestVote to all peers except self */
for (int i = 0; i < s->n_peers; i++) {
if (i == s->self_idx)
continue;
s->cbs->on_send_vote_request(s->ctx, i, s->term);
}
/* Schedule election timeout watchdog */
s->cbs->on_schedule_timeout(s->ctx, s->election_timeout_ms);
return RAFT_OK;
}
int raft_on_vote_request(raft_state_t *s, int from_idx, uint64_t term,
uint64_t *reply_term)
{
/* Step 1: stale term — deny immediately, no state change */
if (term < s->term) {
*reply_term = s->term;
if (s->cbs->on_trace)
s->cbs->on_trace(s->ctx, "vote_denied_stale_term", NULL);
return RAFT_DENY;
}
/* Step 2: term upgrade — update term, clear vote, step down to FOLLOWER */
if (term > s->term) {
s->term = term;
s->voted_for = -1;
s->voted_for_term = 0;
s->role = RAFT_FOLLOWER;
}
/* reply_term reflects our (possibly updated) current term from here on */
*reply_term = s->term;
/* Step 3: role gate — LEADER denies */
if (s->role == RAFT_LEADER) {
if (s->cbs->on_trace)
s->cbs->on_trace(s->ctx, "vote_denied_role", NULL);
return RAFT_DENY;
}
/* Step 4: duplicate-vote deny — already voted for a *different* candidate
* this same term. If voted_for == from_idx and voted_for_term == term,
* fall through to idempotent re-grant path below. */
if (s->voted_for != -1 &&
s->voted_for != from_idx &&
s->voted_for_term == term) {
if (s->cbs->on_trace)
s->cbs->on_trace(s->ctx, "vote_denied_already_voted", NULL);
return RAFT_DENY;
}
/* Step 5: application-level policy gate (freshness, log lsn, etc.) */
if (s->cbs->on_vote_eligible != NULL) {
if (s->cbs->on_vote_eligible(s->ctx, from_idx, term) == 0) {
if (s->cbs->on_trace)
s->cbs->on_trace(s->ctx, "vote_denied_freshness", NULL);
return RAFT_DENY;
}
}
/* Step 6: persist before granting — durability guarantee */
if (s->cbs->on_persist(s->ctx, term, from_idx) != 0) {
/* Persist failed — deny without updating voted_for */
return RAFT_DENY;
}
/* Grant: record vote */
s->voted_for = from_idx;
s->voted_for_term = term;
if (s->cbs->on_trace)
s->cbs->on_trace(s->ctx, "vote_granted", NULL);
return RAFT_GRANT;
}
int raft_on_vote_reply(raft_state_t *s, int from_idx, int granted, uint64_t term)
{
(void)from_idx;
if (s->cbs->on_trace)
s->cbs->on_trace(s->ctx, "vote_reply_in", NULL);
/* Step 1: term upgrade — higher term means we lost the election */
if (term > s->term) {
s->term = term;
s->role = RAFT_FOLLOWER;
s->election_decided = 1;
s->cbs->on_become_follower(s->ctx, s->term, -1);
s->cbs->on_cancel_timeout(s->ctx);
if (s->cbs->on_trace)
s->cbs->on_trace(s->ctx, "became_follower", NULL);
return RAFT_BECAME_FOLLOWER;
}
/* Step 2: no-op guard — election already decided */
if (s->election_decided) {
if (s->cbs->on_trace)
s->cbs->on_trace(s->ctx, "vote_reply_noop", NULL);
return RAFT_PENDING;
}
/* Step 3: count this reply */
s->replies_received++;
/* Step 4: accumulate grant */
if (granted)
s->votes_granted++;
/* Step 5: quorum check */
int quorum = (s->n_peers / 2) + 1;
if (s->votes_granted >= quorum) {
s->role = RAFT_LEADER;
s->election_decided = 1;
s->cbs->on_become_leader(s->ctx, s->term);
s->cbs->on_cancel_timeout(s->ctx);
if (s->cbs->on_trace)
s->cbs->on_trace(s->ctx, "quorum_reached", NULL);
if (s->cbs->on_trace)
s->cbs->on_trace(s->ctx, "became_leader", NULL);
return RAFT_BECAME_LEADER;
}
/* Step 6: all-replied-no-quorum */
if (s->replies_received == s->n_peers - 1 && s->votes_granted < quorum) {
s->role = RAFT_FOLLOWER;
s->election_decided = 1;
s->cbs->on_become_follower(s->ctx, s->term, -1);
if (s->cbs->on_trace)
s->cbs->on_trace(s->ctx, "became_follower", NULL);
return RAFT_BECAME_FOLLOWER;
}
/* Step 7: still waiting */
return RAFT_PENDING;
}
void raft_on_election_timeout(raft_state_t *s)
{
if (s->election_decided) {
/* Req 6.3: cancel any pending timer and be a no-op */
s->cbs->on_cancel_timeout(s->ctx);
if (s->cbs->on_trace)
s->cbs->on_trace(s->ctx, "election_timeout_noop", NULL);
return;
}
s->role = RAFT_FOLLOWER;
s->election_decided = 1;
s->cbs->on_become_follower(s->ctx, s->term, -1);
s->cbs->on_cancel_timeout(s->ctx);
if (s->cbs->on_trace)
s->cbs->on_trace(s->ctx, "election_timeout", NULL);
}
void raft_on_leader_lost(raft_state_t *s)
{
s->voted_for = -1;
s->role = RAFT_FOLLOWER;
s->cbs->on_start_election_after_delay(s->ctx,
(uint32_t)s->my_priority * s->election_timeout_ms);
if (s->cbs->on_trace)
s->cbs->on_trace(s->ctx, "leader_lost", NULL);
}
int raft_check_and_start_election(raft_state_t *s)
{
if (s->voted_for != -1 && s->voted_for != s->self_idx) {
if (s->cbs->on_trace)
s->cbs->on_trace(s->ctx, "jitter_skip", NULL);
return RAFT_SKIP_ELECTION;
}
/* Precondition: raft_on_leader_lost sets role=FOLLOWER, so -EALREADY is unreachable here */
int rc = raft_start_election(s);
(void)rc;
return RAFT_ELECTION_STARTED;
}