forked from lei.chen/myxce
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) <noreply@anthropic.com>
This commit is contained in:
parent
b765fc4705
commit
6d22f2eb1a
3 changed files with 193 additions and 5 deletions
|
|
@ -65,7 +65,7 @@
|
||||||
- _Depends: 4.1_
|
- _Depends: 4.1_
|
||||||
|
|
||||||
- [ ] 5. Integration: transport wiring and end-to-end scenarios
|
- [ ] 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.
|
- 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).
|
- 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_
|
- _Requirements: 7.1_
|
||||||
|
|
|
||||||
|
|
@ -2628,6 +2628,11 @@ void handle_request_vote_reply(ha_state_t *ha,
|
||||||
static int peer_ctrl_msg_handler(void *ep_raw);
|
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 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);
|
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
|
#ifdef XCE_TESTING
|
||||||
/*
|
/*
|
||||||
|
|
@ -2860,6 +2865,79 @@ done:
|
||||||
return 0;
|
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
|
* repl_ack_decode_and_dispatch: decode an inbound packet on a promoted vote
|
||||||
* endpoint while role == CTRL_PRIMARY.
|
* 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.
|
* Routes by ha->role at dispatch time — no explicit handler reassignment needed.
|
||||||
* CTRL_ELECTING → vote_reply_decode_and_dispatch
|
* CTRL_ELECTING → vote_reply_decode_and_dispatch
|
||||||
* CTRL_PRIMARY → repl_ack_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)
|
* 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)
|
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,
|
ctrl_role_t role = (ctrl_role_t)atomic_load_explicit(&ha->role,
|
||||||
memory_order_acquire);
|
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) {
|
switch (role) {
|
||||||
case CTRL_ELECTING:
|
case CTRL_ELECTING:
|
||||||
return vote_reply_decode_and_dispatch(ep, vctx);
|
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);
|
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_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
|
* 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
|
* 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.
|
* 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
|
* Production caller (task 5.1): pre_vote_reply_decode_and_dispatch, reached from
|
||||||
* is the preVoteReply transport dispatch wired in task 5.1. __attribute__((used))
|
* peer_ctrl_msg_handler when a preVoteReply arrives on the outbound PreVote
|
||||||
* keeps the production build clean (no -Wunused-function) until then.
|
* endpoint. Also driven by the XCE_TESTING reply seam.
|
||||||
*/
|
*/
|
||||||
__attribute__((used))
|
|
||||||
static void pre_vote_reply_dispatch(ha_state_t *ha, cluster_t *c,
|
static void pre_vote_reply_dispatch(ha_state_t *ha, cluster_t *c,
|
||||||
int peer_idx, int granted)
|
int peer_idx, int granted)
|
||||||
{
|
{
|
||||||
|
|
|
||||||
|
|
@ -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. */
|
* LOST outcome without a live delaymgr. */
|
||||||
extern int (*ha_prevote_rearm_liveness_hook)(ha_state_t *ha, cluster_t *c);
|
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 ─────────────────────────────────────────────────────────── */
|
/* ── Test config ─────────────────────────────────────────────────────────── */
|
||||||
|
|
||||||
static ha_config_t g_cfg = {
|
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");
|
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 ──────────────────────────────────────────────────── */
|
/* ── Suite entry point ──────────────────────────────────────────────────── */
|
||||||
|
|
||||||
void test_ha_prevote(void)
|
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_round_timeout_is_loss);
|
||||||
RUN_TEST(test_prevote_reusable_after_loss);
|
RUN_TEST(test_prevote_reusable_after_loss);
|
||||||
RUN_TEST(test_prevote_pending_is_noop);
|
RUN_TEST(test_prevote_pending_is_noop);
|
||||||
|
RUN_TEST(test_prevote_reply_over_real_transport_reaches_tally);
|
||||||
UNITY_END();
|
UNITY_END();
|
||||||
}
|
}
|
||||||
|
|
|
||||||
Loading…
Add table
Add a link
Reference in a new issue