wdgzyjx/myxce
1
0
Fork 0
forked from lei.chen/myxce
Code Pull requests Activity

feat(ha-prevote): controller candidate-side PreVote orchestration (task 4.1)

Browse source 
ha.c: raft_cb_send_pre_vote_request (emits preVoteRequest), ha_start_prevote_async
(side-effect-free round + bounded round timeout), pre_vote_reply_dispatch
(raft_on_pre_vote_reply + branch), ha_enter_real_election helper. WON -> truncate
+ CTRL_ELECTING + ha_start_election_async (unchanged real-election path); LOST/
timeout -> stay CTRL_STANDBY + re-arm liveness + log abandonment; PENDING -> no-op.
Round inert until task 4.2 wires triggers; absolute peer-index keyed tally. 6 new
test_ha_prevote tests via reply hook; daemon + suite link clean.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
Rongsong Shen 2026-06-16 05:45:06 +00:00
parent 6f95e4cdfc
commit 7e19bf6110
6 changed files with 801 additions and 1 deletions
Download patch file Download diff file Expand all files Collapse all files

439
src/controller/ha.c
View file

@ -81,6 +81,7 @@ extern int push_config_transport_only(cluster_t *c, const uint8_t *bytes,
#endif
static void on_vote_connected(endpoint_t *ep);
static void on_pre_vote_connected(endpoint_t *ep);
static int vote_ep_err_handler(void *ep_raw);
/* ha-single-leader-role task 2.2: shared promotion seam — every road to
* PRIMARY registers the persist hooks and arms the own-term gate. */
@ -203,6 +204,87 @@ static int raft_cb_send_vote_request(void *ctx, int peer_idx, uint64_t term)
return 0;
}
/*
* raft_cb_send_pre_vote_request: encode and send a capnp PreVote request to a
* peer (ha-prevote task 4.1).
*
* Mirrors raft_cb_send_vote_request exactly — same iomgr connect path, same
* vote_ep_ctx_t carrier, same per-peer endpoint tracking — but the connected
* handler (on_pre_vote_connected) emits OPCode_preVoteRequest instead of
* OPCode_requestVote. prospective_term is the candidate's CURRENT+1 term; it
* is NEVER persisted (PreVote is a side-effect-free straw poll).
*
* peer_idx is the absolute candidate index in c->myconf->peers_cfg.candidates[].
* Returns 0 on success; non-zero in test mode (no cluster) or on alloc failure.
*
* Transport-side reply routing for preVoteReply is wired separately in task 5.1;
* this callback only emits the request.
*/
static int raft_cb_send_pre_vote_request(void *ctx, int peer_idx,
uint64_t prospective_term)
{
ha_state_t *ha = (ha_state_t *)ctx;
cluster_t *c = ha->raft_c;
if (!c || !c->myconf) return 1; /* test mode — caller owns transport */
const xce_cluster_config_t *pcfg = &c->myconf->peers_cfg;
if (peer_idx < 0 || peer_idx >= pcfg->candidates_count)
return 1;
ha_elect_ctx_t *ectx = (ha_elect_ctx_t *)atomic_load_explicit(
&ha->elect_ctx, memory_order_acquire);
if (!ectx) return 1; /* no round metadata carrier */
xce_candidate_t *peer = (xce_candidate_t *)&pcfg->candidates[peer_idx];
/* Map absolute peer_idx → votes[] local index (same convention as the
* real-vote send callback). */
int self_idx = ha->my_priority;
int votes_idx = (peer_idx < self_idx) ? peer_idx : peer_idx - 1;
if (votes_idx < 0 || votes_idx >= ectx->n_peers)
return 1;
vote_ep_ctx_t *vctx = calloc(1, sizeof(*vctx));
if (!vctx) return 1;
vctx->ha = ha;
vctx->c = c;
vctx->ctx = ectx;
/* Carry the ABSOLUTE peer index: the PreVote tally in raft_state_t is keyed
* by absolute cluster index (raft_begin_pre_vote loops over i in
* [0, n_peers) and raft_on_pre_vote_reply dedups by that same index). */
vctx->peer_idx = peer_idx;
vctx->my_term = prospective_term;
vctx->my_tag = (ectx->n_peers > 0) ? ectx->votes[0].my_tag : 0;
vctx->my_lsn = (ectx->n_peers > 0) ? ectx->votes[0].my_lsn : 0;
vctx->my_name = c->myconf ? c->myconf->name : "";
endpoint_t *ep = endpoint_get();
if (!ep) { free(vctx); return 1; }
endpoint_init(ep, AF_INET, SOCK_STREAM, 0, 0);
if (c) endpoint_set_peer(ep, c->myhost);
ep->data = vctx;
ep->iomgr = &c->iomgr;
if (ha->client_ssl_ctx) {
SSL_CTX *client_ctx = (SSL_CTX *)ha->client_ssl_ctx;
ep->ssl = SSL_new(client_ctx);
if (ep->ssl)
SSL_set_connect_state(ep->ssl);
}
endpoint_set_on_connected(ep, on_pre_vote_connected);
endpoint_set_err_handler(ep, vote_ep_err_handler);
int rc = endpoint_connect(ep, peer->hostname, peer->port);
if (rc < 0) { endpoint_fini(ep); return 1; }
rc = iomgr_monitor(ep);
if (rc < 0) { endpoint_fini(ep); return 1; }
return 0;
}
/*
* raft_cb_vote_eligible: freshness gate delegated from raft library.
*
@ -800,6 +882,7 @@ static void raft_cb_trace(void *ctx, const char *event, const char *detail)
static const raft_callbacks_t s_real_cbs = {
.on_send_vote_request = raft_cb_send_vote_request,
.on_send_pre_vote_request = raft_cb_send_pre_vote_request,
.on_vote_eligible = raft_cb_vote_eligible,
.on_become_leader = raft_cb_become_leader,
.on_become_follower = raft_cb_become_follower,
@ -2609,6 +2692,45 @@ static void on_vote_connected(endpoint_t *ep)
ep->msg_handler = peer_ctrl_msg_handler;
}
/*
* on_pre_vote_connected: iomgr callback fired after the TLS handshake completes
* on an outgoing PreVote endpoint (ha-prevote task 4.1).
*
* Mirrors on_vote_connected but emits OPCode_preVoteRequest (reusing the
* RequestVoteRequest payload struct + union arm, per the wire schema in task
* 1.1). prospective_term (current+1) rides in the term field; nothing is
* persisted. The reply-routing handler for preVoteReply is wired in task 5.1;
* here we set the same state-aware dispatcher used by the real-vote path.
*/
static void on_pre_vote_connected(endpoint_t *ep)
{
vote_ep_ctx_t *vctx = (vote_ep_ctx_t *)ep->data;
request_vote_req_t vrq;
memset(&vrq, 0, sizeof(vrq));
vrq.term = vctx->my_term; /* prospective term = current + 1 */
vrq.candidate_id = (char *)(vctx->my_name ? vctx->my_name : "");
vrq.config_tag = vctx->my_tag;
vrq.last_log_lsn = vctx->my_lsn;
xce_request_t req;
memset(&req, 0, sizeof(req));
req.opcode = OPCode_preVoteRequest;
req.body_which = Request_body_requestvotemsg;
req.body.requestvotemsg = &vrq;
#ifdef XCE_TESTING
if (xce_vote_request_hook != NULL)
xce_vote_request_hook(ep, &req);
else
xce_request(ep, &req);
#else
xce_request(ep, &req);
#endif
ep->msg_handler = peer_ctrl_msg_handler;
}
/*
* xce_vote_reply_hook: test-only seam for vote_reply_msg_handler.
*
@ -3320,6 +3442,323 @@ void ha_init_raft(ha_state_t *ha, cluster_t *c, int n_peers)
atomic_load_explicit(&ha->term, memory_order_acquire));
}
/* ── ha-prevote task 4.1: candidate-side PreVote orchestration ─────────────
*
* The PreVote round is a side-effect-free majority probe interposed BEFORE the
* destructive truncate + CTRL_ELECTING + real-election sequence. During the
* round the node stays CTRL_STANDBY and mutates no persistent state (term /
* voted_for / role). Only on a WIN does it enter the unchanged real-election
* path; on a LOSS / round-timeout it stays STANDBY, re-arms the liveness
* watchdog, and logs campaign abandonment.
*
* This task implements the round machinery + the WON/LOST branches. The
* failover triggers (ha_on_primary_ctrl_fd_rdhup / liveness_poll_fn /
* ha_on_heartbeat_received) are rewired to call this in task 4.2 — until then
* the round is INERT in production.
*/
/*
* prevote_round_ctx_t: delaymgr context for the PreVote round timeout.
* cancelled is set when the round is decided (WON or LOST) before the timeout
* fires, so a late timeout task is a no-op.
*/
typedef struct {
_Atomic int cancelled;
ha_state_t *ha;
cluster_t *c;
} prevote_round_ctx_t;
/* Active round-timeout context. _Atomic so arm (vtask) vs timeout-fire
* (delaymgr) vs cancel-on-decision is well-defined. */
static prevote_round_ctx_t * _Atomic s_prevote_round_ctx = NULL;
#ifdef XCE_TESTING
/*
* ha_prevote_rearm_liveness_hook: test-only seam. When non-NULL, the LOST /
* timeout branch calls this INSTEAD OF ha_arm_heartbeat_timeout_vtask so a unit
* test can count re-arms without a live delaymgr (Req 5.4).
*/
int (*ha_prevote_rearm_liveness_hook)(ha_state_t *ha, cluster_t *c) = NULL;
#endif
/* ha_prevote_rearm_liveness: re-arm the standby liveness watchdog after a lost
* round so a later genuine failure still triggers a new PreVote round (the
* watchdog does not re-arm itself once it has fired). */
static void ha_prevote_rearm_liveness(ha_state_t *ha, cluster_t *c)
{
#ifdef XCE_TESTING
if (ha_prevote_rearm_liveness_hook != NULL) {
ha_prevote_rearm_liveness_hook(ha, c);
return;
}
#endif
ha_arm_heartbeat_timeout_vtask(ha, c);
}
/* ha_prevote_cancel_round_timeout: mark the active round-timeout context
* cancelled (the delaymgr task frees it on fire) and clear the static pointer
* so a later decision/timeout sees no active round. */
static void ha_prevote_cancel_round_timeout(void)
{
prevote_round_ctx_t *old = atomic_load_explicit(&s_prevote_round_ctx,
memory_order_acquire);
if (old) {
atomic_store_explicit(&old->cancelled, 1, memory_order_release);
atomic_store_explicit(&s_prevote_round_ctx, NULL, memory_order_release);
}
}
/*
* ha_enter_real_election: the WON-branch sequence — truncate uncommitted WAL
* above the committed boundary, then enter the UNCHANGED real-election path
* (ha_start_election_async transitions to CTRL_ELECTING and delegates the term
* bump + self-vote to raft_start_election). Factored here so both the PreVote
* WON branch (task 4.1) and the rewired trigger (task 4.2) can call it; this
* task does NOT change ha_on_primary_ctrl_fd_rdhup's own behaviour.
*
* c == NULL (unit-test mode): skip truncate + the live election; just transition
* the role to CTRL_ELECTING so the WON transition is observable.
*/
static void ha_enter_real_election(ha_state_t *ha, cluster_t *c)
{
if (!c) {
/* Test mode: the real election needs iomgr/delaymgr; just mark the
* destructive transition so the WON branch is observable. */
atomic_store_explicit(&ha->role, (int32_t)CTRL_ELECTING,
memory_order_release);
return;
}
/* Truncate uncommitted WAL entries above committed_lsn (mirrors the
* existing failover sequence in ha_on_primary_ctrl_fd_rdhup). */
uint64_t committed = atomic_load_explicit(&ha->last_known_committed_lsn,
memory_order_acquire);
if (committed > 0)
persist_truncate_wal_above(&c->persist, committed);
/* Candidate freshness for the RequestVote payload — same floor the voter
* side computes (ha_vote_self_freshness), matching the rdhup path. */
uint64_t cand_tag = 0, cand_lsn = 0;
if (c->myconf) {
uint64_t local_tag = c->myconf->config_version_tag;
uint64_t local_lsn = atomic_load_explicit(&ha->applied_lsn,
memory_order_acquire);
ha_vote_self_freshness(ha, local_tag, local_lsn, &cand_tag, &cand_lsn);
}
int n_peers = (c->myconf) ? c->myconf->peers_cfg.candidates_count : 1;
const char *my_name = (c->myconf) ? c->myconf->name : "";
/* ha_start_election_async transitions to CTRL_ELECTING and drives the
* UNCHANGED raft_start_election (term++ + self-vote). Peer connections are
* driven by raft_cb_send_vote_request; the peers[] arg is unused. */
ha->raft_c = c;
ha_start_election_async(ha, c, n_peers, NULL, cand_tag, cand_lsn, my_name);
}
/*
* ha_start_prevote_async: start a PreVote round (ha-prevote task 4.1).
*
* Snapshots self tag/lsn into the round metadata carrier (ha->elect_ctx, read
* by raft_cb_send_pre_vote_request), calls raft_begin_pre_vote (self-grant +
* per-peer send via the callback), and arms a round-timeout bounded by
* election_timeout_ms (Req 5.4 — no unbounded delay). The node stays
* CTRL_STANDBY; NOTHING persistent is mutated (no term/voted_for/role change).
*
* If raft_begin_pre_vote returns RAFT_PRE_VOTE_WON immediately (single-node
* cluster: self-grant is already a majority), the WON branch runs at once.
*
* c == NULL (unit-test mode): the send callback returns early and the round
* timeout is a no-op; the round state lives entirely in raft_state_t so the
* reply-driven branches are still fully exercisable via the test seam.
*/
void ha_start_prevote_async(ha_state_t *ha, cluster_t *c)
{
if (!ha) return;
ha->raft_c = c;
/* Snapshot self freshness for the PreVote request payload (same floor as a
* real vote). Stored in the metadata carrier ha->elect_ctx so
* raft_cb_send_pre_vote_request can read my_tag/my_lsn. */
uint64_t cand_tag = 0, cand_lsn = 0;
int n_peers = ha->n_peers > 0 ? ha->n_peers : ha->raft.n_peers;
if (c && c->myconf) {
uint64_t local_tag = c->myconf->config_version_tag;
uint64_t local_lsn = atomic_load_explicit(&ha->applied_lsn,
memory_order_acquire);
ha_vote_self_freshness(ha, local_tag, local_lsn, &cand_tag, &cand_lsn);
n_peers = c->myconf->peers_cfg.candidates_count;
int n_other = (n_peers > 1) ? n_peers - 1 : 0;
ha_elect_ctx_t *mc = calloc(1, sizeof(*mc) +
(size_t)n_other * sizeof(mc->votes[0]));
if (mc) {
mc->n_peers = n_other;
mc->result = -1;
mc->c = c;
mc->self_tag = cand_tag;
mc->self_lsn = cand_lsn;
pthread_mutex_init(&mc->lock, NULL);
sem_init(&mc->done, 0, 0);
if (n_other > 0) {
mc->votes[0].my_tag = cand_tag;
mc->votes[0].my_lsn = cand_lsn;
}
/* Do NOT free a prior ctx (live vote endpoints may hold a bare
* pointer to it — same UAF caution as the rdhup path). */
atomic_store_explicit(&ha->elect_ctx, mc, memory_order_release);
}
ha->n_peers = n_peers;
}
INFO("HA: starting PreVote round (prospective term=%" PRIu64 ", n_peers=%d) "
"— staying CTRL_STANDBY, no persist",
ha->raft.term + 1, n_peers);
/* raft_begin_pre_vote: self-grant + per-peer send; mutates only the
* pre_vote_* tally in raft_state_t (no term/voted_for/role change). */
int rc = raft_begin_pre_vote(&ha->raft);
/* Arm the round timeout (bounded by election_timeout_ms). Production only;
* the round state is in raft_state_t so tests exercise the branches via the
* reply seam without the timer. */
ha_prevote_cancel_round_timeout();
if (c && c->delaymgr.running) {
prevote_round_ctx_t *tctx = calloc(1, sizeof(*tctx));
if (tctx) {
atomic_init(&tctx->cancelled, 0);
tctx->ha = ha;
tctx->c = c;
atomic_store_explicit(&s_prevote_round_ctx, tctx,
memory_order_release);
extern int ha_prevote_round_timeout_fn(void *arg);
int trc = delay_task_ms(&c->delaymgr,
(int)ha->config.election_timeout_ms, 1,
(delay_task_entry_fn_t)ha_prevote_round_timeout_fn,
tctx);
if (trc < 0) {
WARN("HA: ha_start_prevote_async: round-timeout schedule failed: %d",
trc);
atomic_store_explicit(&s_prevote_round_ctx, NULL,
memory_order_release);
free(tctx);
}
}
}
/* Single-node: the self-grant is already a majority — win immediately. */
if (rc == RAFT_PRE_VOTE_WON) {
ha_prevote_cancel_round_timeout();
INFO("HA: PreVote won immediately (single-node) — entering real election");
ha_enter_real_election(ha, c);
}
}
/*
* pre_vote_reply_dispatch: feed a PreVote reply into the raft tally and branch
* on its outcome (ha-prevote task 4.1). peer_idx is the ABSOLUTE cluster index
* of the replying peer (the same index raft_begin_pre_vote sent to).
*
* RAFT_PRE_VOTE_WON → cancel round timeout, run the real-election sequence
* (truncate + CTRL_ELECTING + ha_start_election_async).
* RAFT_PRE_VOTE_LOST → stay CTRL_STANDBY, re-arm liveness, log abandonment.
* RAFT_PRE_VOTE_PENDING → no-op (await more replies).
*
* Mirrors vote_reply_dispatch → raft_on_vote_reply, but the raft library makes
* NO role/term change here — the controller acts on the returned status.
*
* In 4.1 the only caller is the XCE_TESTING reply seam; the production caller
* is the preVoteReply transport dispatch wired in task 5.1. __attribute__((used))
* keeps the production build clean (no -Wunused-function) until then.
*/
__attribute__((used))
static void pre_vote_reply_dispatch(ha_state_t *ha, cluster_t *c,
int peer_idx, int granted)
{
int outcome = raft_on_pre_vote_reply(&ha->raft, peer_idx, granted);
if (outcome == RAFT_PRE_VOTE_WON) {
ha_prevote_cancel_round_timeout();
INFO("HA: PreVote round WON — entering real election (term will be %"
PRIu64 ")", ha->raft.term + 1);
ha_enter_real_election(ha, c);
return;
}
if (outcome == RAFT_PRE_VOTE_LOST) {
ha_prevote_cancel_round_timeout();
/* Req 7.3: log campaign abandonment distinctly from a genuine failover
* so operators can tell a suppressed disruptive election apart. */
INFO("HA: PreVote round LOST — abandoning campaign, staying CTRL_STANDBY "
"(no term increment); re-arming liveness watchdog");
/* Stay CTRL_STANDBY — no persist, no role change (Req 1.4). */
ha_prevote_rearm_liveness(ha, c);
return;
}
/* RAFT_PRE_VOTE_PENDING — await more replies, do nothing. */
}
/*
* ha_prevote_round_timeout_fn: delaymgr task body for the PreVote round timeout.
* A timeout that fires before a decision is treated as a LOSS (Req 5.4).
*/
int ha_prevote_round_timeout_fn(void *arg)
{
prevote_round_ctx_t *ctx = (prevote_round_ctx_t *)arg;
if (atomic_load_explicit(&ctx->cancelled, memory_order_acquire)) {
#ifndef XCE_TESTING
free(ctx);
#endif
return 0;
}
ha_state_t *ha = ctx->ha;
cluster_t *c = ctx->c;
/* Only act if the round is still in progress (not already decided). */
if (ha->raft.pre_vote_in_progress) {
ha->raft.pre_vote_in_progress = 0;
atomic_store_explicit(&s_prevote_round_ctx, NULL, memory_order_release);
WARN("HA: PreVote round timed out before a decision — treating as LOSS, "
"staying CTRL_STANDBY; re-arming liveness watchdog (Req 5.4)");
ha_prevote_rearm_liveness(ha, c);
}
#ifndef XCE_TESTING
free(ctx);
#endif
return 0;
}
#ifdef XCE_TESTING
/*
* pre_vote_reply_dispatch_for_test: drive the real pre_vote_reply_dispatch with
* decoded (peer_idx, granted) fields, bypassing iomgr (ha-prevote task 4.1).
*/
void pre_vote_reply_dispatch_for_test(ha_state_t *ha, cluster_t *c,
int peer_idx, int granted)
{
pre_vote_reply_dispatch(ha, c, peer_idx, granted);
}
/*
* ha_prevote_round_timeout_fire_for_test: simulate the round timeout firing
* before a decision, driving the LOST branch (Req 5.4).
*/
void ha_prevote_round_timeout_fire_for_test(ha_state_t *ha, cluster_t *c)
{
prevote_round_ctx_t ctx;
memset(&ctx, 0, sizeof(ctx));
ctx.ha = ha;
ctx.c = c;
atomic_init(&ctx.cancelled, 0);
ha_prevote_round_timeout_fn(&ctx);
}
#endif /* XCE_TESTING */
/* ── Task 6.3: ha_start_election_async — delegate to raft_start_election ────
*
* Non-blocking election initiator. Sets up the raft state machine and