From 6d22f2eb1a110aa71ce0d63affb00a8ec5b32cb1 Mon Sep 17 00:00:00 2001 From: Rongsong Shen Date: Tue, 16 Jun 2026 06:47:08 +0000 Subject: [PATCH] feat(ha-prevote): route preVoteReply over real transport to the candidate tally (task 5.1) peer_ctrl_msg_handler gains a CTRL_STANDBY + pre_vote_in_progress branch that decodes an incoming preVoteReply (no term adoption) and feeds it to pre_vote_reply_dispatch with the absolute peer index, paralleling the real-vote reply path. Gives pre_vote_reply_dispatch its production caller (drops the used attribute). STANDBY-no-round still drops; ELECTING/PRIMARY routing unchanged. PreVote now drives rounds to a decision end-to-end over the wire. Co-Authored-By: Claude Opus 4.8 (1M context) --- .kiro/specs/ha-prevote/tasks.md | 2 +- src/controller/ha.c | 113 ++++++++++++++++++++- src/controller/testsuite/test_ha_prevote.c | 83 +++++++++++++++ 3 files changed, 193 insertions(+), 5 deletions(-) diff --git a/.kiro/specs/ha-prevote/tasks.md b/.kiro/specs/ha-prevote/tasks.md index f157f52d..958ce6b0 100644 --- a/.kiro/specs/ha-prevote/tasks.md +++ b/.kiro/specs/ha-prevote/tasks.md @@ -65,7 +65,7 @@ - _Depends: 4.1_ - [ ] 5. Integration: transport wiring and end-to-end scenarios -- [ ] 5.1 Wire the PreVote reply opcode case to the candidate handler +- [x] 5.1 Wire the PreVote reply opcode case to the candidate handler - Add the transport dispatch case for the incoming PreVote-reply opcode so it routes to the reply handler built in 4.1 (replacing the test hook in production), paralleling the real-vote reply dispatch case. - Observable: a PreVote reply arriving over the real transport reaches the candidate's tally and can drive the round to a decision end-to-end (not just via the test hook). - _Requirements: 7.1_ diff --git a/src/controller/ha.c b/src/controller/ha.c index f2251b0e..49f0d231 100644 --- a/src/controller/ha.c +++ b/src/controller/ha.c @@ -2628,6 +2628,11 @@ void handle_request_vote_reply(ha_state_t *ha, static int peer_ctrl_msg_handler(void *ep_raw); static int vote_reply_decode_and_dispatch(endpoint_t *ep, vote_ep_ctx_t *vctx); static int repl_ack_decode_and_dispatch(endpoint_t *ep, vote_ep_ctx_t *vctx); +/* ha-prevote task 5.1: preVoteReply decode/route; defined alongside the + * real-vote decoder, feeds the later-defined pre_vote_reply_dispatch. */ +static int pre_vote_reply_decode_and_dispatch(endpoint_t *ep, vote_ep_ctx_t *vctx); +static void pre_vote_reply_dispatch(ha_state_t *ha, cluster_t *c, + int peer_idx, int granted); #ifdef XCE_TESTING /* @@ -2860,6 +2865,79 @@ done: return 0; } +/* + * pre_vote_reply_decode_and_dispatch: decode a preVoteReply packet on an + * outbound PreVote endpoint and feed it to the PreVote tally (ha-prevote task + * 5.1). Mirrors vote_reply_decode_and_dispatch exactly — the PreVote reply + * reuses the identical RequestVoteReply payload / union arm — but routes to + * pre_vote_reply_dispatch (the raft-library PreVote tally) instead of the + * real-vote tally, and deliberately performs NO term adoption (a PreVote reply + * never changes our term; Req 1.2). vctx->peer_idx is the ABSOLUTE cluster + * index that raft_on_pre_vote_reply expects (set in ha_send_pre_vote_request). + * Packet ownership: takes ep->fsm_in.pkt and frees it. + */ +static int pre_vote_reply_decode_and_dispatch(endpoint_t *ep, vote_ep_ctx_t *vctx) +{ + ha_state_t *ha = vctx->ha; + cluster_t *c = vctx->c; + int idx = vctx->peer_idx; /* ABSOLUTE cluster index */ + io_packet_t *pkt = ep->fsm_in.pkt; + int ok = 0; + int vr_granted = 0; + + ep->fsm_in.pkt = NULL; /* take ownership (same pattern as + * vote_reply_decode_and_dispatch) */ + + if (!pkt || !pkt->data || pkt->sz == 0) + goto done; + + { + struct capn capn_ctx; + struct Request outer_rq; + Request_ptr outer_rp; + struct RequestReply rep_s; + struct RequestVoteReply vr_raw; + int capn_ok; + + capn_ok = capn_init_mem(&capn_ctx, (const uint8_t *)pkt->data, + (size_t)pkt->sz, 0); + if (capn_ok != 0) + goto done; + + outer_rp.p = capn_getp(capn_root(&capn_ctx), 0, 1); + read_Request(&outer_rq, outer_rp); + + if (outer_rq.body_which != Request_body_replymsg || + !outer_rq.body.replymsg.p.data) { + capn_free(&capn_ctx); + goto done; + } + + read_RequestReply(&rep_s, outer_rq.body.replymsg); + + if (rep_s.data_which != RequestReply_data_requestvotereplymsg) { + capn_free(&capn_ctx); + goto done; + } + + memset(&vr_raw, 0, sizeof(vr_raw)); + read_RequestVoteReply(&vr_raw, rep_s.data.requestvotereplymsg); + + capn_free(&capn_ctx); + + vr_granted = vr_raw.voteGranted ? 1 : 0; + ok = 1; + } + +done: + if (pkt) { free_packet(pkt); free(pkt); } + + /* A malformed / wrong-body reply counts as a non-grant (best-effort + * delivery, per design Event Contract). No term adoption either way. */ + pre_vote_reply_dispatch(ha, c, idx, ok ? vr_granted : 0); + return 0; +} + /* * repl_ack_decode_and_dispatch: decode an inbound packet on a promoted vote * endpoint while role == CTRL_PRIMARY. @@ -3048,7 +3126,15 @@ static int repl_ack_decode_and_dispatch(endpoint_t *ep, vote_ep_ctx_t *vctx) * Routes by ha->role at dispatch time — no explicit handler reassignment needed. * CTRL_ELECTING → vote_reply_decode_and_dispatch * CTRL_PRIMARY → repl_ack_decode_and_dispatch + * CTRL_STANDBY + PreVote → pre_vote_reply_decode_and_dispatch (task 5.1) * any other role → drop packet (WARN) + * + * PreVote note (ha-prevote task 5.1): a preVoteReply arrives on the candidate's + * OUTBOUND PreVote endpoint while the candidate's role is still CTRL_STANDBY + * (a PreVote round performs no role change — Req 1.1). Without this branch the + * reply would hit the default case and be dropped, stranding the round until it + * times out. We gate on raft.pre_vote_in_progress so a STANDBY node with no + * active round still drops unexpected traffic. */ static int peer_ctrl_msg_handler(void *ep_raw) { @@ -3058,6 +3144,11 @@ static int peer_ctrl_msg_handler(void *ep_raw) ctrl_role_t role = (ctrl_role_t)atomic_load_explicit(&ha->role, memory_order_acquire); + + /* PreVote replies land while CTRL_STANDBY with a round in progress. */ + if (role == CTRL_STANDBY && ha->raft.pre_vote_in_progress) + return pre_vote_reply_decode_and_dispatch(ep, vctx); + switch (role) { case CTRL_ELECTING: return vote_reply_decode_and_dispatch(ep, vctx); @@ -3275,6 +3366,21 @@ int peer_ctrl_msg_handler_for_test(endpoint_t *ep, io_packet_t *pkt) return peer_ctrl_msg_handler(ep); } +/* + * pre_vote_reply_msg_handler_for_test: drive a preVoteReply through the REAL + * decode/route path (peer_ctrl_msg_handler) while the candidate is CTRL_STANDBY + * with a PreVote round in progress — the exact production condition under which + * a PreVote reply arrives. Unlike vote_reply_msg_handler_for_test it does NOT + * force the role: it proves the STANDBY+pre_vote_in_progress branch added in + * task 5.1 actually routes the reply to the PreVote tally (instead of dropping + * it). Caller owns ep->data (vote_ep_ctx_t); pkt ownership is transferred. + */ +int pre_vote_reply_msg_handler_for_test(endpoint_t *ep, io_packet_t *pkt) +{ + ep->fsm_in.pkt = pkt; + return peer_ctrl_msg_handler(ep); +} + /* * vote_ep_ctx_set_cluster_for_test: set the cluster_t pointer inside the * vote_ep_ctx_t stored at ep->data. Tests that need the promoted-endpoint @@ -3700,11 +3806,10 @@ void ha_start_prevote_async(ha_state_t *ha, cluster_t *c) * 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. + * Production caller (task 5.1): pre_vote_reply_decode_and_dispatch, reached from + * peer_ctrl_msg_handler when a preVoteReply arrives on the outbound PreVote + * endpoint. Also driven by the XCE_TESTING reply seam. */ -__attribute__((used)) static void pre_vote_reply_dispatch(ha_state_t *ha, cluster_t *c, int peer_idx, int granted) { diff --git a/src/controller/testsuite/test_ha_prevote.c b/src/controller/testsuite/test_ha_prevote.c index 007b7fa2..b2d0211e 100644 --- a/src/controller/testsuite/test_ha_prevote.c +++ b/src/controller/testsuite/test_ha_prevote.c @@ -63,6 +63,33 @@ extern void ha_prevote_round_timeout_fire_for_test(ha_state_t *ha, cluster_t *c) * LOST outcome without a live delaymgr. */ extern int (*ha_prevote_rearm_liveness_hook)(ha_state_t *ha, cluster_t *c); +/* ── Task 5.1 transport-route seams ──────────────────────────────────────── */ + +/* + * pre_vote_reply_msg_handler_for_test: drive a preVoteReply through the REAL + * peer_ctrl_msg_handler decode/route path while CTRL_STANDBY + round in + * progress (the production condition). Proves the reply is NOT dropped and + * reaches pre_vote_reply_dispatch — end-to-end, not via the test hook. + */ +extern int pre_vote_reply_msg_handler_for_test(endpoint_t *ep, io_packet_t *pkt); + +/* vote_reply_build_pkt_for_test: encode a RequestVoteReply capnp packet (the + * preVoteReply reuses the identical payload / union arm). */ +extern io_packet_t *vote_reply_build_pkt_for_test(uint64_t term, int vote_granted, + uint64_t peer_tag, + uint64_t peer_lsn); + +/* vote_ep_ctx_build_for_test: allocate a vote_ep_ctx_t and wire ep->data with + * the ABSOLUTE peer_idx (the index pre_vote_reply_dispatch expects). */ +extern endpoint_t *vote_ep_ctx_build_for_test(endpoint_t *ep, + ha_state_t *ha, + ha_elect_ctx_t *ctx, + int peer_idx, + uint64_t my_term, + uint64_t my_tag, + uint64_t my_lsn, + const char *my_name); + /* ── Test config ─────────────────────────────────────────────────────────── */ static ha_config_t g_cfg = { @@ -323,6 +350,61 @@ static void test_prevote_pending_is_noop(void) printf("test_prevote_pending_is_noop: PASS\n"); } +/* ── Test 7: a preVoteReply over the REAL transport reaches the tally (5.1) ─ */ + +/* + * Task 5.1 (Req 7.1): a preVoteReply arriving on the candidate's outbound + * PreVote endpoint — while the candidate is still CTRL_STANDBY with a round in + * progress — must be ROUTED to the PreVote tally by peer_ctrl_msg_handler (NOT + * dropped because the role is STANDBY), and drive the round to a decision + * end-to-end. This exercises the production decode/route path, not the + * pre_vote_reply_dispatch_for_test hook. + */ +static void test_prevote_reply_over_real_transport_reaches_tally(void) +{ + ha_state_t ha; + TEST_ASSERT_EQUAL_INT(0, ha_init(&ha, &g_cfg, 0)); + ha_do_demote(&ha); /* role = CTRL_STANDBY */ + make_three_node(&ha); + + /* Start a real round: self-grant + in-progress, role stays CTRL_STANDBY. */ + ha_start_prevote_async(&ha, NULL); + TEST_ASSERT_EQUAL_INT_MESSAGE(CTRL_STANDBY, + (int)atomic_load_explicit(&ha.role, memory_order_acquire), + "precondition: role must be CTRL_STANDBY during the round"); + TEST_ASSERT_EQUAL_INT_MESSAGE(1, ha.raft.pre_vote_in_progress, + "precondition: round must be in progress"); + + /* Build an endpoint carrying ABSOLUTE peer_idx = 1 (the index the tally + * dedups by), exactly as ha_send_pre_vote_request sets it. */ + endpoint_t ep; + memset(&ep, 0, sizeof(ep)); + vote_ep_ctx_build_for_test(&ep, &ha, NULL, /*peer_idx=*/1, + /*my_term=*/1, /*my_tag=*/0, /*my_lsn=*/0, "test"); + + /* Encode a GRANT preVoteReply (same wire payload as a real-vote reply). */ + io_packet_t *pkt = vote_reply_build_pkt_for_test(/*term=*/1, /*granted=*/1, + /*tag=*/0, /*lsn=*/0); + TEST_ASSERT_NOT_NULL_MESSAGE(pkt, "vote_reply_build_pkt_for_test must succeed"); + + /* Route it through the REAL handler (peer_ctrl_msg_handler) while STANDBY. */ + int rc = pre_vote_reply_msg_handler_for_test(&ep, pkt); + TEST_ASSERT_EQUAL_INT_MESSAGE(0, rc, "handler must return 0"); + + /* Peer grant → quorum (2 of 3) → WON: the reply was NOT dropped; it reached + * the tally and drove the round to a decision (real-election sequence). */ + int role = (int)atomic_load_explicit(&ha.role, memory_order_acquire); + TEST_ASSERT_TRUE_MESSAGE(role == CTRL_ELECTING || role == CTRL_PRIMARY, + "Req 7.1: a preVoteReply over the real transport must reach the tally " + "and drive the round to a decision (left CTRL_STANDBY)"); + TEST_ASSERT_EQUAL_INT_MESSAGE(0, ha.raft.pre_vote_in_progress, + "the decided round must clear the in-progress flag"); + + free(ep.data); /* vctx allocated by vote_ep_ctx_build_for_test */ + ha_fini(&ha); + printf("test_prevote_reply_over_real_transport_reaches_tally: PASS (role=%d)\n", role); +} + /* ── Suite entry point ──────────────────────────────────────────────────── */ void test_ha_prevote(void) @@ -334,5 +416,6 @@ void test_ha_prevote(void) RUN_TEST(test_prevote_round_timeout_is_loss); RUN_TEST(test_prevote_reusable_after_loss); RUN_TEST(test_prevote_pending_is_noop); + RUN_TEST(test_prevote_reply_over_real_transport_reaches_tally); UNITY_END(); }