From fd9933c9d0f474e29d7818fe0f5de3772a9a81cd Mon Sep 17 00:00:00 2001 From: Rongsong Shen Date: Sun, 31 May 2026 08:43:01 +0000 Subject: [PATCH] =?UTF-8?q?feat(raft-library):=20task=206.3=20=E2=80=94=20?= =?UTF-8?q?delegate=20election=20dispatch=20to=20raft=5F*=20API=20in=20ha.?= =?UTF-8?q?c=20+=20cluster.c?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .kiro/specs/raft-library/tasks.md | 2 +- src/controller/cluster.c | 66 ++++- src/controller/ha.c | 387 +++++++++--------------------- 3 files changed, 163 insertions(+), 292 deletions(-) diff --git a/.kiro/specs/raft-library/tasks.md b/.kiro/specs/raft-library/tasks.md index a53ae2e7..c216466c 100644 --- a/.kiro/specs/raft-library/tasks.md +++ b/.kiro/specs/raft-library/tasks.md @@ -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)` diff --git a/src/controller/cluster.c b/src/controller/cluster.c index b086b115..70afde2a 100644 --- a/src/controller/cluster.c +++ b/src/controller/cluster.c @@ -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; diff --git a/src/controller/ha.c b/src/controller/ha.c index c035c8ed..8b9ee7d0 100644 --- a/src/controller/ha.c +++ b/src/controller/ha.c @@ -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 ─────────────────────────────────────────── */