lei.chen/myxce
1
0
Fork 1
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

fix(ha): election-safety hardening — vote gates, restored-vote term, freshness source

Browse source 
Election protocol fixes validated against raft.tla invariants:

- ha_init_raft(): one-time raft state init at startup (re-init would lose
  durable election progress); n_peers = full cluster size including self
- handle_request_vote(): higher-term step-down moved BEFORE the role gate
  (split-brain fix); standby vote-deny now requires a live primary control
  channel (primary_ctrl_fd >= 0) AND a fresh heartbeat
- ha_heartbeat_is_fresh(): derive freshness from ha->last_activity (the
  field production actually writes); the orphaned last_heartbeat timespec
  had no production writer, leaving the vote-deny gate inert (bug-337);
  field removed from ha_state_t
- handle_request_vote(): restore same-candidate idempotent re-grant —
  deny only a DIFFERENT candidate in the same term, matching raft.c and
  Raft §5.2 votedFor ∈ {Nil, candidateId} (bug-338)
- raft_init(): restore voted_for_term alongside a persisted vote; leaving
  it 0 allowed a double vote in the same term across a restart (bug-339)
- myapp_run(): leader_known check treats a vote recorded under this node's
  own name as a self-vote — production persists the real candidate name,
  the "self" placeholder only appears in test mode
- raft_on_vote_request(): leader steps down on higher-term vote request
  instead of denying it

Tests: test_proto_H6_restored_vote_blocks_second_vote added; H1 restore
assertion strengthened; test_ha_vote_gate.c driven through last_activity +
primary_ctrl_fd (the real production fields). Full /xce-build all green
(lib + controller + scheduler + Go suites, 0 failures, ASAN clean).

deploy-ha.sh: start.sh now forwards extra args to the controller binary.
This commit is contained in:
Rongsong Shen 2026-06-10 01:25:39 +00:00
parent fa66d1ba9c
commit 9821ec4f3c
7 changed files with 207 additions and 91 deletions
Download patch file Download diff file Expand all files Collapse all files

119
src/controller/ha.c
View file

@ -1753,21 +1753,23 @@ int ha_restore_term(ha_state_t *ha, const char *workdir)
/* ── ha_heartbeat_is_fresh (Req 6.2) ────────────────────────────────────────
*
* Returns 1 if the elapsed time since ha->last_heartbeat is strictly less than
* ha->config.heartbeat_timeout_ms (i.e. the primary is still alive).
* Returns 0 if the heartbeat is stale (timeout has expired).
* Returns 1 if the elapsed time since the last primary message
* (ha->last_activity, written by ha_on_heartbeat_received) is strictly less
* than ha->config.heartbeat_timeout_ms (i.e. the primary is still alive).
* Returns 0 if no primary message has been seen yet (last_activity == 0) or
* the timeout has expired.
*
* Uses CLOCK_MONOTONIC to avoid wall-clock skew.
* Uses CLOCK_MONOTONIC seconds to avoid wall-clock skew; whole-second
* granularity is sufficient for the 10 s default timeout.
*/
static int ha_heartbeat_is_fresh(ha_state_t *ha)
{
struct timespec now, last;
time_t last = ha->last_activity; /* time_t load is atomic on LP64 */
if (last == 0)
return 0; /* no primary message seen yet */
struct timespec now;
clock_gettime(CLOCK_MONOTONIC, &now);
pthread_mutex_lock(&ha->lock);
last = ha->last_heartbeat;
pthread_mutex_unlock(&ha->lock);
long ms = (now.tv_sec - last.tv_sec) * 1000L
+ (now.tv_nsec - last.tv_nsec) / 1000000L;
long ms = (long)(now.tv_sec - last) * 1000L;
return ms < (long)ha->config.heartbeat_timeout_ms;
}
@ -1837,8 +1839,9 @@ int handle_request_vote(ha_state_t *ha,
memory_order_acquire);
if (my_role == CTRL_PRIMARY || my_role == CTRL_PRIMARY_COMMITTING)
return 0; /* deny — I am primary (or committing); protect against challengers */
if (my_role == CTRL_STANDBY && ha_heartbeat_is_fresh(ha))
return 0; /* deny — my primary is alive */
if (my_role == CTRL_STANDBY && ha->primary_ctrl_fd >= 0 &&
ha_heartbeat_is_fresh(ha))
return 0; /* deny — my primary is alive (active primary connection) */
/* Req 2.5: catching-up standbys deny all votes */
if (in_catch_up_flag)
@ -1847,9 +1850,10 @@ int handle_request_vote(ha_state_t *ha,
/* Protect voted_for check+set from concurrent vote requests on different vtasks */
pthread_mutex_lock(&ha->lock);
/* Req 2.6: already voted for a different candidate in this term → deny */
if (ha->voted_for_term == term &&
ha->voted_for_name[0] != '\0' &&
/* Req 2.6: already voted for a DIFFERENT candidate in this term → deny.
* Same-candidate re-grant is idempotent (Raft §5.2: grant iff votedFor ∈
* {Nil, candidateId}) — matches the duplicate-vote check in raft.c. */
if (ha->voted_for_term == term && ha->voted_for_name[0] != '\0' &&
strcmp(ha->voted_for_name, candidate_id) != 0) {
pthread_mutex_unlock(&ha->lock);
return 0;
@ -2975,6 +2979,49 @@ static int vote_ep_err_handler(void *ep_raw)
return 0;
}
/* ── ha_init_raft: one-time raft state machine initialization ────────────────
*
* Called once at startup after ha_restore_term() to wire the raft state
* machine with the correct cluster size and callbacks. Subsequent elections
* use raft_start_election() / raft_on_leader_lost() on this same state.
*
* n_peers: total candidates_count from cluster config (includes self).
*/
void ha_init_raft(ha_state_t *ha, cluster_t *c, int n_peers)
{
/* Map persisted voted_for_name back to a peer index for initial_voted_for.
* This lets raft_check_and_start_election (RDHUP path) see the prior vote
* without re-reading the file. */
int restored_voted_for = -1;
if (ha->voted_for_name[0] != '\0' && c && c->myconf) {
const xce_cluster_config_t *pcfg = &c->myconf->peers_cfg;
for (int i = 0; i < pcfg->candidates_count; i++) {
if (strcmp(pcfg->candidates[i].name, ha->voted_for_name) == 0) {
restored_voted_for = i;
break;
}
}
}
raft_config_t rcfg;
memset(&rcfg, 0, sizeof(rcfg));
rcfg.n_peers = n_peers;
rcfg.self_idx = ha->my_priority;
rcfg.election_timeout_ms = ha->config.election_timeout_ms;
rcfg.my_priority = ha->my_priority;
rcfg.initial_term = atomic_load_explicit(&ha->term, memory_order_acquire);
rcfg.initial_voted_for = restored_voted_for;
rcfg.cbs = &s_real_cbs;
rcfg.ctx = ha;
raft_init(&ha->raft, &rcfg);
ha->n_peers = n_peers;
ha->raft_c = c;
INFO("HA: raft initialized (n_peers=%d, self=%d, term=%" PRIu64 ")",
n_peers, ha->my_priority,
atomic_load_explicit(&ha->term, memory_order_acquire));
}
/* ── Task 6.3: ha_start_election_async — delegate to raft_start_election ────
*
* Non-blocking election initiator. Sets up the raft state machine and
@ -3001,45 +3048,8 @@ void ha_start_election_async(ha_state_t *ha, cluster_t *c,
(void)peers; /* peer connections driven by raft_cb_send_vote_request */
(void)my_name; /* self-vote name stored in on_persist via raft_start_election */
/* Step 0: wire cluster pointer for raft callbacks.
* Must be set before any raft_* call so callbacks can reach cluster_t. */
ha->raft_c = c;
/* Step 0b: re-initialise raft_state_t with the real cluster size and
* the term/voted_for restored from disk by ha_restore_term().
*
* ha->raft.term may be 0 from ha_init while ha->term was restored to a
* higher value by ha_restore_term() — reconcile here before the election.
*
* self_idx = ha->my_priority (index of this node in candidates[]).
* n_peers = total cluster size (candidates_count); passed directly.
*
* voted_for: map ha->voted_for_name back to a peer index (-1 if none).
*/
{
int restored_voted_for = -1;
if (ha->voted_for_name[0] != '\0' && c && c->myconf) {
const xce_cluster_config_t *pcfg = &c->myconf->peers_cfg;
for (int i = 0; i < pcfg->candidates_count; i++) {
if (strcmp(pcfg->candidates[i].name, ha->voted_for_name) == 0) {
restored_voted_for = i;
break;
}
}
}
raft_config_t rcfg;
memset(&rcfg, 0, sizeof(rcfg));
rcfg.n_peers = n_peers; /* total candidates_count */
rcfg.self_idx = ha->my_priority;
rcfg.election_timeout_ms = ha->config.election_timeout_ms;
rcfg.my_priority = ha->my_priority;
rcfg.initial_term = atomic_load_explicit(&ha->term, memory_order_acquire);
rcfg.initial_voted_for = restored_voted_for;
rcfg.cbs = &s_real_cbs;
rcfg.ctx = ha;
raft_init(&ha->raft, &rcfg);
}
/* Step 0: raft is initialized once at startup by ha_init_raft(); nothing
* to do here. ha->raft_c and ha->n_peers are already set. */
/* Step 1: transition to CTRL_ELECTING before firing RPCs.
* raft_start_election fires on_become_leader / on_send_vote_request below;
@ -3255,6 +3265,7 @@ void ha_on_primary_ctrl_fd_rdhup(ha_state_t *ha, cluster_t *c)
/* Update n_peers tracking — total cluster size (including self) */
ha->n_peers = n_peers;
}
}
#ifdef XCE_TESTING

19
src/controller/ha.h
View file

@ -151,7 +151,7 @@ typedef struct {
_Atomic uint64_t applied_lsn; /* Standby: last applied WAL LSN */
char primary_host[XCE_MAXNAME_LEN]; /* Hostname of current primary */
uint32_t primary_port; /* Port of current primary */
struct timespec last_heartbeat; /* CLOCK_MONOTONIC timestamp of last heartbeat */
/* last_heartbeat deleted — heartbeat freshness now derives from last_activity */
_Atomic int promotion_failed;
_Atomic int promotion_retry_count;
_Atomic int in_catch_up; /* Non-zero while processing historical WAL entries */
@ -159,7 +159,7 @@ typedef struct {
int hooks_registered; /* Idempotency guard for xce_add_hook calls */
uint32_t last_received_crc; /* Chained CRC of last received replicated entry */
ha_config_t config;
pthread_mutex_t lock; /* Protects primary_host/port + last_heartbeat */
pthread_mutex_t lock; /* Protects primary_host/port + voted_for_term/name */
/* watchdog_stop / watchdog_thread deleted — replaced by ha_arm_heartbeat_timeout_vtask */
void *ssl_ctx; /* SSL_CTX* cast; avoids pulling openssl headers here */
uint64_t jitter_seed; /* xorshift64 PRNG state for election jitter */
@ -441,6 +441,21 @@ int ha_handle_snapshot_chunk(ha_state_t *ha, persist_t *p,
*/
void ha_persist_term(ha_state_t *ha, const char *workdir);
/*
* ha_init_raft: initialize the embedded raft state machine with cluster topology.
*
* Must be called ONCE at controller startup after config is loaded and
* ha_restore_term() has run. Subsequent elections (initial and re-elections
* after primary loss) reuse the same raft state — only raft_start_election()
* and raft_on_leader_lost() are called for each election cycle.
*
* Calling raft_init() more than once would reset raft's internal state
* (term, voted_for, role) and could lose durable election progress.
*
* n_peers: total cluster size including self (candidates_count from config).
*/
void ha_init_raft(ha_state_t *ha, cluster_t *c, int n_peers);
/*
* ha_restore_term: read {workdir}/ha-term.bin at startup and restore
* ha->term and ha->voted_for_* into ha_state_t.

33
src/controller/myapp.c
View file

@ -217,6 +217,11 @@ int myapp_run(myconf_t *myconf)
uint64_t my_lsn = (uint64_t)atomic_load_explicit(&c.ha.applied_lsn,
memory_order_acquire);
/* Initialize raft state machine once with cluster topology.
* Must happen before any election or watchdog path that calls
* raft_on_leader_lost() / raft_start_election(). */
ha_init_raft(&c.ha, &c, n_peers);
/* Startup stagger: delay election start by priority * election_priority_step_ms.
* Node 0 starts first; lower-priority nodes wait, see the winner's vote request
* arrive, and grant the vote instead of starting their own election.
@ -240,11 +245,24 @@ int myapp_run(myconf_t *myconf)
* path (iomgr was already running); the raft term + primary_ctrl_fd checks
* catch the case where iomgr was NOT running during the nap (connections
* queued in the kernel backlog) but the winner's ctrlhello has already
* established the control channel. */
* established the control channel.
*
* voted_for_name is combined with term: a self-vote means this node was
* a candidate, not a follower — no external leader is known. In
* production raft_cb_persist records the self-vote under this node's own
* name (myconf->name); the "self" placeholder only appears in test mode,
* so both spellings must be checked. A stale vote (voted_for_term <
* ha->term) is also discarded. All these cases must fall through to
* ha_start_election_async. */
{
int leader_known = 0;
pthread_mutex_lock(&c.ha.lock);
if (c.ha.voted_for_name[0] != '\0')
uint64_t vft = c.ha.voted_for_term;
uint64_t cur = atomic_load_explicit(&c.ha.term, memory_order_relaxed);
if (c.ha.voted_for_name[0] != '\0' &&
strcmp(c.ha.voted_for_name, "self") != 0 &&
strcmp(c.ha.voted_for_name, myconf->name) != 0 &&
vft == cur)
leader_known = 1;
pthread_mutex_unlock(&c.ha.lock);
if (!leader_known && c.ha.raft.term > 0)
@ -253,10 +271,10 @@ int myapp_run(myconf_t *myconf)
leader_known = 1;
if (leader_known) {
INFO("myapp_run: leader already known after startup delay "
"(voted_for=%s, raft_term=%" PRIu64 ", primary_fd=%d) — "
"skipping own election",
"(voted_for=%s, voted_term=%" PRIu64 ", raft_term=%" PRIu64
", primary_fd=%d) — skipping own election",
c.ha.voted_for_name[0] ? c.ha.voted_for_name : "(none)",
c.ha.raft.term, c.ha.primary_ctrl_fd);
vft, c.ha.raft.term, c.ha.primary_ctrl_fd);
goto skip_election;
}
}
@ -281,6 +299,11 @@ skip_election:;
atomic_store_explicit(&c.ha.role, (int32_t)CTRL_STANDBY,
memory_order_release);
INFO("myapp_run: CTRL_STANDBY — waiting for primary to connect and push config");
/* Arm heartbeat watchdog even when we skipped the election.
* If the known primary never connects within heartbeat_timeout_ms,
* the watchdog fires ha_on_primary_ctrl_fd_rdhup which increments
* the term (clearing voted_for) and starts a fresh election. */
ha_arm_heartbeat_timeout_vtask(&c.ha, &c);
}
}

36
src/controller/testsuite/test_ha_vote_gate.c
View file

@ -47,24 +47,16 @@ static ha_config_t g_gate_cfg = {
.election_priority_step_ms = 500,
};
/* ── Helper: set ha->last_heartbeat to "now - offset_ms milliseconds ago" ── */
/* ── Helper: set ha->last_activity to "now - offset_ms milliseconds ago" ──
*
* last_activity is the production field written by ha_on_heartbeat_received
* (whole CLOCK_MONOTONIC seconds); offset_ms is rounded down to seconds. */
static void set_last_heartbeat_age(ha_state_t *ha, long offset_ms)
static void set_last_activity_age(ha_state_t *ha, long offset_ms)
{
struct timespec now;
clock_gettime(CLOCK_MONOTONIC, &now);
long sec = offset_ms / 1000;
long nsec = (offset_ms % 1000) * 1000000L;
ha->last_heartbeat.tv_sec = now.tv_sec - sec;
ha->last_heartbeat.tv_nsec = now.tv_nsec - nsec;
/* Carry / borrow normalisation */
if (ha->last_heartbeat.tv_nsec < 0) {
ha->last_heartbeat.tv_sec -= 1;
ha->last_heartbeat.tv_nsec += 1000000000L;
}
ha->last_activity = now.tv_sec - offset_ms / 1000;
}
/* ── Req 6.1: CTRL_PRIMARY → deny regardless of candidate freshness ────── */
@ -139,8 +131,13 @@ static void test_standby_fresh_heartbeat_denies_vote_request(void)
atomic_store_explicit(&ha.term, 2ULL, memory_order_release);
ha.voted_for_term = 0;
/* last_heartbeat = 500 ms ago (< 10 000 ms timeout → fresh) */
set_last_heartbeat_age(&ha, 500L);
/* Active primary control channel (handle_ctrlhello sets this on a real
* standby; the deny gate requires a live connection AND a fresh heartbeat).
* ha_fini does not close the fd, so any non-negative value is safe. */
ha.primary_ctrl_fd = 999;
/* last_activity = 500 ms ago → rounds to "this second" (< 10 s → fresh) */
set_last_activity_age(&ha, 500L);
int granted = handle_request_vote(&ha,
2ULL,
@ -174,8 +171,11 @@ static void test_standby_stale_heartbeat_evaluates_vote(void)
atomic_store_explicit(&ha.term, 1ULL, memory_order_release);
ha.voted_for_term = 0;
/* last_heartbeat = 15 000 ms ago (> 10 000 ms timeout → stale) */
set_last_heartbeat_age(&ha, 15000L);
/* Active primary control channel — staleness alone must open the gate */
ha.primary_ctrl_fd = 999;
/* last_activity = 15 000 ms ago (> 10 000 ms timeout → stale) */
set_last_activity_age(&ha, 15000L);
int granted = handle_request_vote(&ha,
1ULL,

39
src/lib/raft/raft.c
View file

@ -40,7 +40,11 @@ int raft_init(raft_state_t *s, const raft_config_t *cfg)
s->term = cfg->initial_term;
s->role = RAFT_FOLLOWER;
s->voted_for = cfg->initial_voted_for;
s->voted_for_term = 0;
/* A restored vote belongs to the restored term: persistence writes
* (term, voted_for) as a pair. Leaving voted_for_term at 0 would let the
* duplicate-vote check miss after restart, granting a second vote in the
* same term (Election Safety violation). */
s->voted_for_term = (cfg->initial_voted_for != -1) ? cfg->initial_term : 0;
s->n_peers = cfg->n_peers;
s->self_idx = cfg->self_idx;
s->votes_granted = 0;
@ -132,22 +136,35 @@ int raft_on_vote_request(raft_state_t *s, int from_idx, uint64_t term,
/* reply_term reflects our (possibly updated) current term from here on */
*reply_term = s->term;
/* Step 3: role gate — LEADER denies */
/* Step 3: role gate — same-term challenger denied by live leader.
*
* This only fires when term == s->term (the equal-term case). For
* term > s->term, Step 2 already downgraded s->role to RAFT_FOLLOWER
* before we reach here, so the check below is false and we fall through
* to grant/deny based on freshness.
*
* A RequestVote for the same term as the current leader means the sender
* started a competing candidacy in a term the leader already won — deny. */
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 4: duplicate-vote check — already voted in this term.
* Idempotent re-grant: if this is the same candidate we already voted for
* in this term, fall through so we re-persist and re-grant (network retry
* path). This is safe: the candidate has already committed to this term
* and we already durably recorded the vote.
* Different candidate: deny — standard Raft invariant of at most one vote
* per node per term. */
if (s->voted_for != -1 && s->voted_for_term == term) {
if (s->voted_for != from_idx) {
if (s->cbs->on_trace)
s->cbs->on_trace(s->ctx, "vote_denied_already_voted", NULL);
return RAFT_DENY;
}
/* same candidate — fall through to re-persist and re-grant */
}
/* Step 5: application-level policy gate (freshness, log lsn, etc.) */

49
src/lib/testsuite/test_raft.c
View file

@ -2029,6 +2029,9 @@ static void test_proto_H1_init_with_persisted_state(void)
TEST_ASSERT_EQUAL_INT(0, rc);
TEST_ASSERT_EQUAL_UINT64(5, s.term);
TEST_ASSERT_EQUAL_INT(1, s.voted_for);
/* The restored vote belongs to the restored term — without this, the
* duplicate-vote check cannot match after restart (see H6). */
TEST_ASSERT_EQUAL_UINT64(5, s.voted_for_term);
TEST_ASSERT_EQUAL_INT(RAFT_FOLLOWER, s.role);
}
@ -2061,6 +2064,51 @@ static void test_proto_H2_init_first_boot(void)
TEST_ASSERT_EQUAL_INT(RAFT_FOLLOWER, s.role);
}
/*
* test_proto_H6_restored_vote_blocks_second_vote
*
* Election Safety across restart: a node that voted for candidate 1 in
* term 5, then restarted (raft_init with initial_term=5, initial_voted_for=1),
* must DENY a vote request from a DIFFERENT candidate (idx=2) in the same
* term 5. Without restoring voted_for_term alongside voted_for, the
* duplicate-vote check cannot match the restored term and the node would
* grant a second vote in term 5 → two leaders in one term.
*/
static void test_proto_H6_restored_vote_blocks_second_vote(void)
{
raft_state_t s;
raft_callbacks_t cbs = make_mock_cbs();
uint64_t reply_term = 0;
reset_mock();
raft_config_t cfg;
memset(&s, 0, sizeof(s));
memset(&cfg, 0, sizeof(cfg));
cfg.n_peers = 3;
cfg.self_idx = 0;
cfg.election_timeout_ms = 1000;
cfg.initial_term = 5;
cfg.initial_voted_for = 1; /* persisted vote for candidate 1, term 5 */
cfg.cbs = &cbs;
TEST_ASSERT_EQUAL_INT(0, raft_init(&s, &cfg));
/* Different candidate (idx=2) requests a vote in the SAME term 5 */
int rc = raft_on_vote_request(&s, 2, 5, &reply_term);
TEST_ASSERT_EQUAL_INT(RAFT_DENY, rc);
TEST_ASSERT_EQUAL_UINT64(5, reply_term);
/* The restored vote must be untouched */
TEST_ASSERT_EQUAL_INT(1, s.voted_for);
TEST_ASSERT_EQUAL_UINT64(5, s.voted_for_term);
/* Same candidate (idx=1) retransmitting must still be re-granted */
rc = raft_on_vote_request(&s, 1, 5, &reply_term);
TEST_ASSERT_EQUAL_INT(RAFT_GRANT, rc);
}
/*
* test_proto_H3_persist_before_vote_recorded
*
@ -2964,6 +3012,7 @@ void test_raft(void)
RUN_TEST(test_proto_H3_persist_before_vote_recorded);
RUN_TEST(test_proto_H4_persist_fail_denies);
RUN_TEST(test_proto_H5_start_election_self_vote_persisted);
RUN_TEST(test_proto_H6_restored_vote_blocks_second_vote);
RUN_TEST(test_proto_Hnull_required_callback_einval);
/* Group D: Quorum and Result (task 5.3) */

3
src/tools/deploy-ha.sh
View file

@ -52,12 +52,13 @@ if [ -z "\$NAME" ]; then
echo "Usage: \$0 <ctrl-name>"
exit 1
fi
shift
CTRL_DIR=$TOP/controller/\$NAME
if [ ! -d "\$CTRL_DIR/conf" ] || [ ! -d "\$CTRL_DIR/data" ]; then
echo "Error: \$CTRL_DIR/conf or \$CTRL_DIR/data does not exist"
exit 1
fi
$BINARY --myname \$NAME --peers "$PEERS_STR" --workdir $TOP/controller --certdir $TOP/certs --logfile $TOP/logs/\$NAME.log --vcpus 4 \$VERBOSE
$BINARY --myname \$NAME --peers "$PEERS_STR" --workdir $TOP/controller --certdir $TOP/certs --logfile $TOP/logs/\$NAME.log --vcpus 4 \$*
EOF
chmod +x "$TOP/scripts/start.sh"