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

feat(raft-library): task 6.3 — delegate election dispatch to raft_* API in ha.c + cluster.c

Browse source
This commit is contained in:
Rongsong Shen 2026-05-31 08:43:01 +00:00
parent 8f0d2d55c3
commit fd9933c9d0
3 changed files with 163 additions and 292 deletions
Download patch file Download diff file Expand all files Collapse all files

2
.kiro/specs/raft-library/tasks.md
View file

@ -160,7 +160,7 @@
- _Requirements: 9.1_
- _Boundary: ha.c callbacks_
- [ ] 6.3 Delegate election state management to raft_* API
- [x] 6.3 Delegate election state management to raft_* API
- Remove direct `ha->term++` and `ha->voted_for = ...` writes from `ha_start_election_async`; replace with `raft_start_election(&ha->raft)`
- Replace vote request handling in the capnp vote-request handler with `raft_on_vote_request(&ha->raft, from_idx, term, &reply_term)`
- Replace vote reply accumulation with `raft_on_vote_reply(&ha->raft, from_idx, granted, term)`

66
src/controller/cluster.c
View file

@ -258,23 +258,63 @@ static int myreq_handler(vtask_t* task)
rc = do_clr_preason(c, ep, &request);
break;
case OPCode_requestVote: {
/* Task 6.3: delegate vote-request handling to raft library.
*
* The freshness gate (config_tag / log_lsn comparison) is applied here
* BEFORE calling raft_on_vote_request because raft_cb_vote_eligible
* (on_vote_eligible) intentionally skips the freshness check — it
* relies on the caller (this dispatcher) to enforce it first.
*
* raft_on_vote_request handles:
* - stale-term deny (term < current_term)
* - term upgrade (term > current_term → clear voted_for)
* - role gate: RAFT_LEADER denies; COMMITTING gate in on_vote_eligible
* - one-vote-per-term invariant (voted_for != -1 && voted_for != from_idx)
* - on_persist before granting (durability guarantee)
*
* reply_term is set by raft_on_vote_request to the current term
* (after any term upgrade), so vr_rep.term reflects the latest term. */
request_vote_req_t *vrq = request.body.requestvotemsg;
uint64_t cand_tag = vrq ? vrq->config_tag : 0;
uint64_t cand_lsn = vrq ? vrq->last_log_lsn : 0;
uint64_t cand_term = vrq ? vrq->term : 0;
uint64_t my_tag = c->myconf->config_version_tag;
uint64_t my_lsn = (uint64_t)atomic_load_explicit(
&c->ha.applied_lsn, memory_order_acquire);
int granted = handle_request_vote(
&c->ha, cand_term, vrq ? vrq->candidate_id : "",
cand_tag, cand_lsn, my_tag, my_lsn,
0 /* in_catch_up_flag: not implemented here */);
uint64_t cand_tag = vrq ? vrq->config_tag : 0;
uint64_t cand_lsn = vrq ? vrq->last_log_lsn : 0;
uint64_t cand_term = vrq ? vrq->term : 0;
uint64_t my_tag = c->myconf->config_version_tag;
uint64_t my_lsn = (uint64_t)atomic_load_explicit(
&c->ha.applied_lsn, memory_order_acquire);
/* Freshness gate: deny if candidate is not at least as fresh as us.
* (Req 2.4: config_tag must be ≥ ours; if equal, log_lsn must be ≥ ours.) */
int freshness_ok = !(cand_tag < my_tag ||
(cand_tag == my_tag && cand_lsn < my_lsn));
/* Map candidate name to integer index for raft_on_vote_request. */
int from_idx = -1;
const char *cand_id = vrq ? vrq->candidate_id : "";
if (cand_id && cand_id[0] != '\0') {
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, cand_id) == 0) {
from_idx = i;
break;
}
}
}
uint64_t reply_term = 0;
int granted = 0;
if (freshness_ok && from_idx >= 0) {
int r = raft_on_vote_request(&c->ha.raft, from_idx, cand_term,
&reply_term);
granted = (r == RAFT_GRANT) ? 1 : 0;
} else {
/* Freshness deny or unknown candidate: read current term for reply */
reply_term = atomic_load_explicit(&c->ha.term, memory_order_acquire);
}
request_vote_reply_t vr_rep;
memset(&vr_rep, 0, sizeof(vr_rep));
vr_rep.term = (uint64_t)atomic_load_explicit(
&c->ha.term, memory_order_acquire);
vr_rep.vote_granted = (granted > 0) ? 1 : 0;
vr_rep.term = reply_term;
vr_rep.vote_granted = granted;
vr_rep.peer_config_tag = my_tag;
vr_rep.peer_last_log_lsn = my_lsn;

387
src/controller/ha.c
View file

@ -1877,8 +1877,12 @@ int (*xce_vote_reply_hook)(ha_state_t *ha, ha_elect_ctx_t *ctx, int peer_idx,
#endif /* XCE_TESTING */
/*
* vote_reply_dispatch: internal helper — calls handle_request_vote_reply or
* vote_reply_dispatch: internal helper — delegates to the raft library or
* the test hook (when XCE_TESTING and hook is non-NULL).
*
* Production path: raft_on_vote_reply owns quorum tracking and fires
* on_become_leader/on_become_follower via the raft callbacks.
* Test path: xce_vote_reply_hook captures decoded fields without real iomgr.
*/
static void vote_reply_dispatch(ha_state_t *ha, ha_elect_ctx_t *ctx, int idx,
uint64_t term, int vote_granted,
@ -1890,7 +1894,8 @@ static void vote_reply_dispatch(ha_state_t *ha, ha_elect_ctx_t *ctx, int idx,
return;
}
#endif
handle_request_vote_reply(ha, ctx, idx, term, vote_granted, peer_tag, peer_lsn);
(void)ctx; (void)peer_tag; (void)peer_lsn; /* consumed by raft library via callbacks */
raft_on_vote_reply(&ha->raft, idx, vote_granted, term);
}
/*
@ -1983,11 +1988,17 @@ done:
if (pkt) { free_packet(pkt); free(pkt); }
if (ok) {
/* Term fencing: if peer's term exceeds ours, update atomically */
uint64_t my_term = atomic_load_explicit(&ha->term, memory_order_acquire);
if (vr_term > my_term)
atomic_store_explicit(&ha->term, vr_term, memory_order_release);
/* Term fencing: if peer's term > ours, atomically advance ha->term.
* Must be done here (before vote_reply_dispatch) so it applies on both
* the production path (raft_on_vote_reply) and the XCE_TESTING hook path
* where raft_cb_become_follower is bypassed. */
uint64_t cur = atomic_load_explicit(&ha->term, memory_order_acquire);
while (vr_term > cur) {
if (atomic_compare_exchange_weak_explicit(&ha->term, &cur, vr_term,
memory_order_acq_rel,
memory_order_acquire))
break;
}
vote_reply_dispatch(ha, ctx, idx, vr_term, vr_granted, vr_tag, vr_lsn);
} else {
/* Decode error: treat peer as denied */
@ -2250,32 +2261,33 @@ static int vote_ep_err_handler(void *ep_raw)
return 0;
}
/* ── Task 2.4: ha_start_election_async ─────────────────────────────────────
/* ── Task 6.3: ha_start_election_async — delegate to raft_start_election ────
*
* Non-blocking election initiator. Returns immediately after registering
* all outgoing vote endpoints with iomgr and scheduling the timeout watchdog.
* Non-blocking election initiator. Sets up the raft state machine and
* delegates all election mechanics to raft_start_election().
*
* Single-node shortcircuit (n_peers == 1):
* Allocates a self-only ha_elect_ctx_t, calls ha_elect_evaluate_and_signal
* directly, frees ctx. No iomgr or delaymgr involvement.
* raft_start_election() handles:
* - term increment and self-vote (via on_persist callback)
* - single-node shortcircuit (fires on_become_leader immediately)
* - multi-node RPC fan-out (via on_send_vote_request callback per peer)
* - election timeout scheduling (via on_schedule_timeout callback)
*
* Multi-node (n_peers > 1):
* For each peer (peers[0..n_peers-2]): allocate vote_ep_ctx_t, get a fresh
* endpoint_t, init with client_ssl_ctx, set on_vote_connected +
* vote_ep_err_handler, call endpoint_connect + iomgr_monitor. Skip peer
* on any failure (peer counted as denied via err_handler path later, or
* immediately if connect returns synchronous failure and err_handler fires).
* Allocate election_timeout_ctx_t (cancelled=0) and schedule
* delay_task_ms(c->delaymgr, election_timeout_ms, 1, election_timeout_fn, t).
* ha_elect_ctx_t is still allocated here because raft_cb_send_vote_request
* reads ha->elect_ctx to retrieve my_tag/my_lsn for the capnp VoteRequest
* payload. Vote counting and quorum decisions are now owned by the raft
* library (raft_on_vote_reply fires on_become_leader/follower callbacks).
*
* Requirements: 1.3, 2.1 (shortcircuit), 2.4 (async vote send)
* Requirements: 1.3, 2.1 (shortcircuit), 2.4 (async vote send), 9.2
*/
void ha_start_election_async(ha_state_t *ha, cluster_t *c,
int n_peers, xce_candidate_t **peers,
uint64_t my_config_tag, uint64_t my_lsn,
const char *my_name)
{
/* Step 0: wire cluster pointer for raft callbacks (task 6.2).
(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;
@ -2286,10 +2298,7 @@ void ha_start_election_async(ha_state_t *ha, cluster_t *c,
* 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); ha_start_election_async
* receives n_peers as the count EXCLUDING self, so total = n_peers+0
* is wrong — n_peers already IS the total candidates_count when
* called from cluster.c. Use it directly for raft_state_t.n_peers.
* n_peers = total cluster size (candidates_count); passed directly.
*
* voted_for: map ha->voted_for_name back to a peer index (-1 if none).
*/
@ -2318,228 +2327,76 @@ void ha_start_election_async(ha_state_t *ha, cluster_t *c,
raft_init(&ha->raft, &rcfg);
}
/* Step 1: store peer count */
/* Step 1: store peer count for ha_repl_ack_received quorum calculation */
ha->n_peers = n_peers;
/* Increment term for this election round */
atomic_fetch_add_explicit(&ha->term, 1, memory_order_acq_rel);
uint64_t cur_term = atomic_load_explicit(&ha->term, memory_order_acquire);
/* Record self-vote before sending RequestVote RPCs. This prevents granting
* a vote to a concurrent candidate in the same term (split-brain guard). */
if (my_name) {
pthread_mutex_lock(&ha->lock);
ha->voted_for_term = cur_term;
strncpy(ha->voted_for_name, my_name, sizeof(ha->voted_for_name) - 1);
ha->voted_for_name[sizeof(ha->voted_for_name) - 1] = '\0';
pthread_mutex_unlock(&ha->lock);
}
/* Transition to CTRL_ELECTING */
/* Step 2: transition to CTRL_ELECTING before firing RPCs.
* raft_start_election fires on_become_leader / on_send_vote_request below;
* the role must already be ELECTING so ha_dispatch_gate() admits vote RPCs. */
atomic_store_explicit(&ha->role, (int32_t)CTRL_ELECTING, memory_order_release);
INFO("HA FSM: → CTRL_ELECTING (async, term=%" PRIu64 ", n_peers=%d)",
cur_term, n_peers);
INFO("HA FSM: → CTRL_ELECTING (async, n_peers=%d)", n_peers);
/* Step 2: single-node shortcircuit — self-elect immediately */
if (n_peers == 1) {
ha_elect_ctx_t *ctx = calloc(1, sizeof(*ctx));
if (!ctx) {
WARN("ha_start_election_async: calloc ctx failed (single-node)");
return;
}
ctx->n_peers = 0; /* no other peers */
ctx->votes_granted = 1; /* self-vote */
ctx->n_replied = 0;
ctx->max_resp_tag = 0;
ctx->max_resp_lsn = 0;
ctx->result = -1;
ctx->done_posted = 0;
ctx->c = c; /* Task 4.1: pass cluster for phase2 + committing timeout */
pthread_mutex_init(&ctx->lock, NULL);
sem_init(&ctx->done, 0, 0);
/* Store candidate own tag/lsn in ctx fields (votes[] has 0 entries
* for n_peers==0 shortcircuit — votes[0] would be out-of-bounds). */
ctx->max_resp_tag = my_config_tag;
ctx->max_resp_lsn = my_lsn;
/* For single-node, quorum = ceil(1/2) = 1; votes_granted=1 → wins.
* Task 4.1: ha_elect_evaluate_and_signal now handles COMMITTING entry,
* phase2, repl_ack_count reset, and committing timeout arming. */
INFO("HA election: outcome=PRIMARY term=%" PRIu64
" votes_granted=1 quorum=1 (single-node async)", cur_term);
/* evaluate sets ha->role = CTRL_PRIMARY_COMMITTING and calls phase2 */
ha_elect_evaluate_and_signal(ha, ctx);
pthread_mutex_destroy(&ctx->lock);
sem_destroy(&ctx->done);
free(ctx);
return;
}
/* Step 3: multi-node — n_other = n_peers - 1 peers to solicit */
int n_other = n_peers - 1;
/* Allocate shared election context */
/* Step 3: pre-allocate ha_elect_ctx_t so raft_cb_send_vote_request can
* read my_tag/my_lsn from it when building the capnp VoteRequest payload.
* Vote counting is now owned by the raft library; this ctx is only a
* metadata carrier for tag/lsn. For multi-node elections it is kept alive
* until all per-peer vote_ep_ctx_t endpoints finish (iomgr lifetime).
* For single-node, raft_start_election fires on_become_leader immediately
* so no ctx reads occur after return. */
int n_other = (n_peers > 1) ? n_peers - 1 : 0;
ha_elect_ctx_t *ctx = calloc(1, sizeof(*ctx) +
(size_t)n_other * sizeof(ctx->votes[0]));
if (!ctx) {
WARN("ha_start_election_async: calloc ctx failed (multi-node)");
WARN("ha_start_election_async: calloc ctx failed");
return;
}
ctx->n_peers = n_other;
ctx->votes_granted = 1; /* self-vote */
ctx->votes_granted = 1; /* self-vote placeholder (raft tracks actual count) */
ctx->n_replied = 0;
ctx->max_resp_tag = 0;
ctx->max_resp_lsn = 0;
ctx->result = -1;
ctx->done_posted = 0;
ctx->c = c; /* Task 4.1: pass cluster for phase2 + committing timeout */
ctx->c = c;
pthread_mutex_init(&ctx->lock, NULL);
sem_init(&ctx->done, 0, 0);
/* Store candidate's own tag/lsn in votes[0] for the yield gate */
/* Store candidate's own tag/lsn; votes[0] is read by raft_cb_send_vote_request */
if (n_other > 0) {
ctx->votes[0].my_tag = my_config_tag;
ctx->votes[0].my_lsn = my_lsn;
}
/* Store ctx in ha for cleanup path */
/* Store ctx in ha so raft_cb_send_vote_request can find it via ha->elect_ctx */
atomic_store_explicit(&ha->elect_ctx, ctx, memory_order_release);
/* Connect to each peer and register with iomgr */
int peers_connected = 0;
for (int i = 0; peers != NULL && i < n_other; i++) {
xce_candidate_t *peer = peers[i];
if (!peer) continue;
/* Step 4: delegate election to raft library.
*
* raft_start_election:
* - increments ha->raft.term and records self-vote (fires on_persist)
* - single-node: fires on_become_leader → raft_cb_become_leader
* - multi-node: fires on_send_vote_request per peer → raft_cb_send_vote_request
* then fires on_schedule_timeout → raft_cb_schedule_timeout
*
* No direct ha->term++ or ha->voted_for_* writes happen in this function
* after this call; the raft library owns those via on_persist. */
raft_start_election(&ha->raft);
/* Allocate per-peer context */
vote_ep_ctx_t *vctx = calloc(1, sizeof(*vctx));
if (!vctx) {
WARN("ha_start_election_async: calloc vctx failed for peer %d", i);
/* Count as denied immediately */
handle_request_vote_reply(ha, ctx, i, cur_term, 0, 0, 0);
continue;
}
vctx->ha = ha;
vctx->c = c;
vctx->ctx = ctx;
vctx->peer_idx = i;
vctx->my_term = cur_term;
vctx->my_tag = my_config_tag;
vctx->my_lsn = my_lsn;
vctx->my_name = my_name;
/* Get a fresh endpoint from the iomgr pool */
endpoint_t *ep = endpoint_get();
if (!ep) {
WARN("ha_start_election_async: endpoint_get() failed for peer %d", i);
free(vctx);
handle_request_vote_reply(ha, ctx, i, cur_term, 0, 0, 0);
continue;
}
/* Initialize endpoint: AF_INET/SOCK_STREAM client socket for outgoing vote */
endpoint_init(ep, AF_INET, SOCK_STREAM, 0, 0);
if (c) endpoint_set_peer(ep, c->myhost);
ep->data = vctx;
if (c) ep->iomgr = &c->iomgr;
/* Pre-create client SSL BEFORE endpoint_connect so netio uses it directly.
* endpoint_connect skips SSL_new when ep->ssl is already set (Req 2.2). */
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);
}
/* Set callbacks */
endpoint_set_on_connected(ep, on_vote_connected);
endpoint_set_err_handler(ep, vote_ep_err_handler);
/* Connect (blocking TCP + async SSL via iomgr after) */
int rc = endpoint_connect(ep, peer->hostname, peer->port);
if (rc < 0) {
WARN("ha_start_election_async: endpoint_connect failed for peer %d (%s:%d): %d",
i, peer->hostname, peer->port, rc);
/* endpoint_fini frees ep->data (vctx) and ep->ssl — do not double-free */
endpoint_fini(ep);
/* In production mode (c != NULL), let the election timeout handle unreachable
* peers — do NOT count as immediate denial, or a startup race (ECONNREFUSED
* before peers are listening) collapses the election before the timeout fires.
* In test mode (c == NULL), count as denied so the election can resolve. */
if (!c)
handle_request_vote_reply(ha, ctx, i, cur_term, 0, 0, 0);
continue;
}
/* Register with iomgr for async I/O */
if (c) {
rc = iomgr_monitor(ep);
if (rc < 0) {
WARN("ha_start_election_async: iomgr_monitor failed for peer %d: %d", i, rc);
free(vctx);
endpoint_fini(ep);
handle_request_vote_reply(ha, ctx, i, cur_term, 0, 0, 0);
continue;
}
}
peers_connected++;
INFO("ha_start_election_async: peer %d (%s:%d) connecting async",
i, peer->hostname, peer->port);
}
/* Step 4: schedule election timeout watchdog */
if (c) {
election_timeout_ctx_t *t = calloc(1, sizeof(*t));
if (t) {
atomic_init(&t->cancelled, 0);
t->ha = ha;
t->c = c;
t->ctx = ctx;
int rc = delay_task_ms(&c->delaymgr,
(int)ha->config.election_timeout_ms,
1,
election_timeout_fn,
t);
if (rc < 0) {
WARN("ha_start_election_async: delay_task_ms failed: %d", rc);
free(t);
/* Fallback: evaluate now if no peers connected */
if (peers_connected == 0) {
pthread_mutex_lock(&ctx->lock);
ha_elect_evaluate_and_signal(ha, ctx);
pthread_mutex_unlock(&ctx->lock);
}
} else {
/* Fix 4 (HIGH): wire timeout_ctx back into ctx so that task 4.1
* (quorum-reached path in handle_request_vote_reply) can cancel
* the watchdog by setting timeout_ctx->cancelled = 1 atomically. */
ctx->timeout_ctx = t;
}
} else {
WARN("ha_start_election_async: calloc election_timeout_ctx_t failed");
}
} else if (peers_connected == 0 && n_other > 0) {
/* No iomgr and no peers connected (test mode with n_peers>1, peers=NULL):
* evaluate immediately so the election doesn't stall, then free ctx
* since no delaymgr task owns it. */
pthread_mutex_lock(&ctx->lock);
ha_elect_evaluate_and_signal(ha, ctx);
pthread_mutex_unlock(&ctx->lock);
/* For single-node, raft_start_election fires on_become_leader synchronously.
* on_become_leader (raft_cb_become_leader) handles role → COMMITTING,
* phase2, and committing timeout. Free the ctx here since no raft callbacks
* will reference it after raft_start_election returns for single-node. */
if (n_peers == 1) {
atomic_store_explicit(&ha->elect_ctx, NULL, memory_order_release);
pthread_mutex_destroy(&ctx->lock);
sem_destroy(&ctx->done);
free(ctx);
ctx = NULL;
return;
}
INFO("ha_start_election_async: returning (term=%" PRIu64
", n_peers=%d, connected=%d)", cur_term, n_peers, peers_connected);
/* Returns immediately — election result delivered via ha_elect_evaluate_and_signal */
INFO("ha_start_election_async: returning (n_peers=%d, raft election started)",
n_peers);
/* Returns immediately — election result delivered via raft on_become_leader/follower */
}
/* ── Task 9.2: Persistent Control Channel ──────────────────────────────────
*
@ -2744,72 +2601,46 @@ void ha_on_primary_ctrl_fd_rdhup(ha_state_t *ha, cluster_t *c)
if (committed > 0)
persist_truncate_wal_above(&c->persist, committed);
/* Clear voted_for: the old primary is dead; any vote from a previous term
* is stale and must not block the new election. */
pthread_mutex_lock(&ha->lock);
ha->voted_for_term = 0;
memset(ha->voted_for_name, 0, sizeof(ha->voted_for_name));
pthread_mutex_unlock(&ha->lock);
/* Task 6.3: delegate RDHUP election trigger to raft library.
*
* raft_on_leader_lost:
* - clears ha->raft.voted_for = -1 (no direct ha->voted_for_* writes here)
* - sets ha->raft.role = RAFT_FOLLOWER
* - fires on_start_election_after_delay(ctx, priority * election_timeout_ms)
* → raft_cb_start_election_after_delay schedules raft_rdhup_delay_fn
* which calls raft_check_and_start_election after the priority jitter.
*
* The manual nanosleep jitter + voted_for clear + skip-if-already-voted
* logic is now fully owned by the raft state machine.
*
* raft->raft_c must be set before raft_on_leader_lost fires callbacks. */
ha->raft_c = c;
/* Priority-based jitter before RDHUP election.
* The highest-priority (index 0) node starts immediately; lower-priority
* nodes sleep for (priority * election_timeout_ms). During the sleep they
* may receive a RequestVote from the priority-0 node — if so, they grant
* it (setting voted_for) and then skip their own election below. */
if (ha->my_priority > 0) {
uint32_t step_ms = ha->config.election_timeout_ms
? ha->config.election_timeout_ms
: HA_DEFAULT_ELECTION_TIMEOUT_MS;
long delay_ns = (long)ha->my_priority * (long)step_ms * 1000000L;
struct timespec jitter = {
.tv_sec = delay_ns / 1000000000L,
.tv_nsec = delay_ns % 1000000000L,
};
nanosleep(&jitter, NULL);
/* Skip our election if we already voted for ANOTHER candidate in the
* current term during the jitter — meaning the priority-0 node sent
* us a RequestVote, we granted it, and should not compete. */
pthread_mutex_lock(&ha->lock);
uint64_t cur_term = atomic_load_explicit(&ha->term, memory_order_acquire);
const char *myname_ptr = c && c->myconf ? c->myconf->name : "";
int already_voted = (ha->voted_for_name[0] != '\0' &&
ha->voted_for_term == cur_term &&
strcmp(ha->voted_for_name, myname_ptr) != 0);
pthread_mutex_unlock(&ha->lock);
if (already_voted) {
INFO("HA: RDHUP — already voted during jitter, skipping own election");
return;
}
}
/* Fire async election — ha_start_election_async is non-blocking (Req 6.3, 6.5).
* ha->phase2_done==1 at this point (set after initial startup election), so
* ha_elect_evaluate_and_signal will skip cluster_init_phase2 and go directly
* to CTRL_PRIMARY_COMMITTING (Req 9.3). */
if (c->myconf && c->myconf->peers_cfg.candidates_count > 0) {
const xce_cluster_config_t *cfg = &c->myconf->peers_cfg;
int n = cfg->candidates_count > 0 ? cfg->candidates_count - 1 : 0;
/* Build peer pointer array (exclude self) */
xce_candidate_t *peer_ptrs[MAX_HA_CANDIDATES];
int np = 0;
for (int i = 0; i < cfg->candidates_count && np < n; i++) {
if (strcmp(cfg->candidates[i].name, c->myhost ? c->myhost->name : "") != 0)
peer_ptrs[np++] = &cfg->candidates[i];
}
uint64_t my_tag = c->myconf->config_version_tag;
uint64_t my_lsn = atomic_load_explicit(&c->persist.committed_lsn,
memory_order_acquire);
const char *my_name = c->myconf ? c->myconf->name : "";
#ifdef XCE_TESTING
if (ha_rdhup_election_hook != NULL) {
if (ha_rdhup_election_hook != NULL) {
/* Test seam: build peer args for hook (same as before) so existing
* tests that capture ha_start_election_async arguments still work. */
if (c->myconf && c->myconf->peers_cfg.candidates_count > 0) {
const xce_cluster_config_t *cfg = &c->myconf->peers_cfg;
int n = cfg->candidates_count > 0 ? cfg->candidates_count - 1 : 0;
xce_candidate_t *peer_ptrs[MAX_HA_CANDIDATES];
int np = 0;
for (int i = 0; i < cfg->candidates_count && np < n; i++) {
if (strcmp(cfg->candidates[i].name,
c->myhost ? c->myhost->name : "") != 0)
peer_ptrs[np++] = &cfg->candidates[i];
}
uint64_t my_tag = c->myconf->config_version_tag;
uint64_t my_lsn = atomic_load_explicit(&c->persist.committed_lsn,
memory_order_acquire);
const char *my_name = c->myconf ? c->myconf->name : "";
ha_rdhup_election_hook(ha, c, np, peer_ptrs, my_tag, my_lsn, my_name);
} else {
ha_start_election_async(ha, c, np, peer_ptrs, my_tag, my_lsn, my_name);
}
#else
ha_start_election_async(ha, c, np, peer_ptrs, my_tag, my_lsn, my_name);
#endif
return;
}
#endif
raft_on_leader_lost(&ha->raft);
}
/* ── Heartbeat timeout watchdog ─────────────────────────────────────────── */