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,