feat(election-iomgr-refactor): implement vote_reply_msg_handler

Replace stub with full iomgr-native vote reply decoder:
- Reads RequestVoteReply fields into local scalars BEFORE capn_free (UAF fix)
- Term fencing: atomic update ha->term if peer term > ours
- Error path: decode failure counts peer as denied vote
- Test seam: xce_vote_reply_hook for unit testing without real packets
- 4 new ASAN-clean unit tests: field decode, denied vote, term fencing, no-UAF

The pre-existing vote_request_thread UAF in the old pthread path remains
but is targeted for removal in task 5.3 (old code removal).
This commit is contained in:
Rongsong Shen 2026-05-30 08:45:43 +00:00
parent 18c280429e
commit bd58e5045a
2 changed files with 464 additions and 6 deletions

View file

@ -1119,18 +1119,232 @@ static void on_vote_connected(endpoint_t *ep)
}
/*
* vote_reply_msg_handler: stub — implemented in task 2.3.
* Declared here so on_vote_connected can reference it.
* Signature matches msg_handler_t = int (*)(void*) from netio.h.
* xce_vote_reply_hook: test-only seam for vote_reply_msg_handler.
*
* Forward-declared here (before vote_reply_msg_handler) so the handler body
* can reference it. Defined only under XCE_TESTING; guarded below.
*
* When non-NULL, vote_reply_msg_handler() calls this INSTEAD OF the real
* handle_request_vote_reply() so tests can capture decoded fields without
* a live iomgr. Receives the decoded scalar fields — no capnp pointers.
*
* Pattern mirrors xce_vote_request_hook used in task 2.2.
*/
static int vote_reply_msg_handler(void *ep)
#ifdef XCE_TESTING
int (*xce_vote_reply_hook)(ha_state_t *ha, ha_elect_ctx_t *ctx, int peer_idx,
uint64_t term, int vote_granted,
uint64_t peer_tag, uint64_t peer_lsn) = NULL;
#endif /* XCE_TESTING */
/*
* vote_reply_dispatch: internal helper — calls handle_request_vote_reply or
* the test hook (when XCE_TESTING and hook is non-NULL).
*/
static void vote_reply_dispatch(ha_state_t *ha, ha_elect_ctx_t *ctx, int idx,
uint64_t term, int vote_granted,
uint64_t peer_tag, uint64_t peer_lsn)
{
(void)ep;
/* Task 2.3 will implement this; stub returns 0 to avoid crash if called */
#ifdef XCE_TESTING
if (xce_vote_reply_hook != NULL) {
xce_vote_reply_hook(ha, ctx, idx, term, vote_granted, peer_tag, peer_lsn);
return;
}
#endif
handle_request_vote_reply(ha, ctx, idx, term, vote_granted, peer_tag, peer_lsn);
}
/*
* vote_reply_msg_handler: iomgr message handler for incoming RequestVoteReply.
*
* Called by iomgr (via INVOKE_MSG_HANDLER) when a full packet arrives on an
* outgoing vote endpoint after on_vote_connected fired.
*
* Steps:
* 1. Cast ep to endpoint_t*; get packet from ep->fsm_in.pkt (owned by us).
* 2. Raw capnp decode: init capn_ctx, read outer Request wrapper.
* 3. read_RequestReply() into rep_s (capnp pointer — still valid).
* 4. read_RequestVoteReply() into vr_raw (scalars) BEFORE capn_free.
* 5. capn_free() — capnp segment released; vr_raw still valid (scalars only).
* 6. free_packet() + free(pkt) — packet memory released.
* 7. Term fencing: if vr_raw.term > ha->term, update ha->term atomically.
* 8. Call vote_reply_dispatch() with vr_raw scalar fields.
*
* On any decode error: treat peer as denied (grant=0, tag=0, lsn=0).
*
* ASAN safety: all capnp fields read before capn_free() — no heap-use-after-free.
* Requirements: 3.1 (iomgr packet), 3.2 (no capnp UAF), 3.3 (extract fields),
* 3.4 (term fencing).
*/
static int vote_reply_msg_handler(void *ep_raw)
{
endpoint_t *ep = (endpoint_t *)ep_raw;
vote_ep_ctx_t *vctx = (vote_ep_ctx_t *)ep->data;
ha_state_t *ha = vctx->ha;
ha_elect_ctx_t *ctx = vctx->ctx;
int idx = vctx->peer_idx;
io_packet_t *pkt = ep->fsm_in.pkt; /* ownership transferred by INVOKE_MSG_HANDLER */
int ok = 0;
/* Decoded reply fields — populated on success path */
uint64_t vr_term = 0;
int vr_granted = 0;
uint64_t vr_tag = 0;
uint64_t vr_lsn = 0;
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;
}
/* CRITICAL: copy all scalars BEFORE capn_free to avoid UAF.
* struct RequestVoteReply fields: term, voteGranted, peerConfigTag,
* peerLastLogLsn — all scalars; safe to copy before freeing segment. */
memset(&vr_raw, 0, sizeof(vr_raw));
read_RequestVoteReply(&vr_raw, rep_s.data.requestvotereplymsg);
capn_free(&capn_ctx); /* capnp segment released; vr_raw.* still valid */
vr_term = (uint64_t)vr_raw.term;
vr_granted = vr_raw.voteGranted ? 1 : 0;
vr_tag = (uint64_t)vr_raw.peerConfigTag;
vr_lsn = (uint64_t)vr_raw.peerLastLogLsn;
ok = 1;
}
done:
/* Release packet — INVOKE_MSG_HANDLER will null ep->fsm_in.pkt after return */
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);
vote_reply_dispatch(ha, ctx, idx, vr_term, vr_granted, vr_tag, vr_lsn);
} else {
/* Decode error: treat peer as denied */
vote_reply_dispatch(ha, ctx, idx,
atomic_load_explicit(&ha->term, memory_order_acquire),
0, 0, 0);
}
return 0;
}
#ifdef XCE_TESTING
/*
* vote_reply_build_pkt_for_test: test helper that encodes a RequestVoteReply
* into a newly allocated io_packet_t that vote_reply_msg_handler can decode.
*
* Wire format mirrors the real peer reply:
* Request{opcode=reply, body.replymsg=RequestReply{
* data_which=requestvotereplymsg,
* data.requestvotereplymsg=RequestVoteReply{term, voteGranted, ...}
* }}
*
* Returns a freshly allocated io_packet_t* on success, NULL on failure.
* Handler takes ownership: it calls free_packet()+free() when done.
* Tests set ep->fsm_in.pkt to the returned pointer before calling the handler.
*/
io_packet_t *vote_reply_build_pkt_for_test(uint64_t term, int vote_granted,
uint64_t peer_tag, uint64_t peer_lsn)
{
struct capn cs;
struct capn_segment *seg;
struct RequestVoteReply vr_capn;
RequestVoteReply_ptr vr_ptr;
struct RequestReply rr_capn;
RequestReply_ptr rr_ptr;
struct Request rq_capn;
Request_ptr rq_ptr;
io_packet_t *pkt;
int rc;
capn_init_malloc(&cs);
seg = capn_root(&cs).seg;
/* RequestVoteReply — all scalar fields */
memset(&vr_capn, 0, sizeof(vr_capn));
vr_capn.term = term;
vr_capn.voteGranted = vote_granted ? 1 : 0;
vr_capn.peerConfigTag = peer_tag;
vr_capn.peerLastLogLsn = peer_lsn;
vr_ptr = new_RequestVoteReply(seg);
write_RequestVoteReply(&vr_capn, vr_ptr);
/* RequestReply envelope */
memset(&rr_capn, 0, sizeof(rr_capn));
rr_capn.seq = 1;
rr_capn.opcode = OPCode_requestVote;
rr_capn.status = 0;
rr_capn.data_which = RequestReply_data_requestvotereplymsg;
rr_capn.data.requestvotereplymsg = vr_ptr;
rr_ptr = new_RequestReply(seg);
write_RequestReply(&rr_capn, rr_ptr);
/* Outer Request{opcode=reply} */
memset(&rq_capn, 0, sizeof(rq_capn));
rq_capn.opcode = OPCode_reply;
rq_capn.seq = 1;
rq_capn.body_which = Request_body_replymsg;
rq_capn.body.replymsg = rr_ptr;
rq_ptr = new_Request(seg);
write_Request(&rq_capn, rq_ptr);
capn_setp(capn_root(&cs), 0, rq_ptr.p);
pkt = calloc(1, sizeof(*pkt));
if (!pkt) { capn_free(&cs); return NULL; }
pkt->sz = capn_size(&cs);
rc = alloc_packet(pkt);
if (rc < 0) { capn_free(&cs); free(pkt); return NULL; }
capn_write_mem(&cs, (uint8_t *)pkt->data, (size_t)pkt->sz, 0);
capn_free(&cs);
return pkt;
}
/*
* vote_reply_msg_handler_for_test: expose vote_reply_msg_handler under
* XCE_TESTING so tests can call it directly (it is otherwise static).
* Sets ep->fsm_in.pkt = pkt then calls vote_reply_msg_handler(ep).
* Caller owns ep->data (vote_ep_ctx_t); pkt ownership is transferred.
*/
int vote_reply_msg_handler_for_test(endpoint_t *ep, io_packet_t *pkt)
{
ep->fsm_in.pkt = pkt;
return vote_reply_msg_handler(ep);
}
/*
* vote_ep_ctx_build_for_test: test helper that allocates a vote_ep_ctx_t,
* fills it from the supplied parameters, and wires up ep->data and

View file

@ -318,6 +318,245 @@ static void test_election_timeout_ctx_cancelled_flag(void)
printf("test_election_timeout_ctx_cancelled_flag: PASS\n");
}
/* ── Task 2.3: vote_reply_msg_handler tests ─────────────────────────────────
*
* Test seam: xce_vote_reply_hook captures decoded fields without real iomgr.
* vote_reply_build_pkt_for_test encodes a real capnp RequestVoteReply packet.
* vote_reply_msg_handler_for_test sets ep->fsm_in.pkt and calls the handler.
*
* ASAN: handler calls free_packet()+free(pkt) before accessing vr_raw fields,
* so there is no heap-use-after-free under AddressSanitizer.
*/
/* Forward declarations for task 2.3 test seams (defined in ha.c XCE_TESTING) */
extern int (*xce_vote_reply_hook)(ha_state_t *ha, ha_elect_ctx_t *ctx,
int peer_idx, uint64_t term, int vote_granted,
uint64_t peer_tag, uint64_t peer_lsn);
extern io_packet_t *vote_reply_build_pkt_for_test(uint64_t term, int vote_granted,
uint64_t peer_tag,
uint64_t peer_lsn);
extern int vote_reply_msg_handler_for_test(endpoint_t *ep, io_packet_t *pkt);
/* ── Mock vote_reply_hook ───────────────────────────────────────────────── */
static int g_reply_hook_calls = 0;
static uint64_t g_reply_term = 0;
static int g_reply_granted = -1;
static uint64_t g_reply_peer_tag = 0;
static uint64_t g_reply_peer_lsn = 0;
static int g_reply_peer_idx = -1;
static int mock_vote_reply_hook(ha_state_t *ha, ha_elect_ctx_t *ctx,
int peer_idx, uint64_t term, int vote_granted,
uint64_t peer_tag, uint64_t peer_lsn)
{
(void)ha; (void)ctx;
g_reply_hook_calls++;
g_reply_peer_idx = peer_idx;
g_reply_term = term;
g_reply_granted = vote_granted;
g_reply_peer_tag = peer_tag;
g_reply_peer_lsn = peer_lsn;
return 0;
}
static void reset_reply_mock(void)
{
g_reply_hook_calls = 0;
g_reply_term = 0;
g_reply_granted = -1;
g_reply_peer_tag = 0;
g_reply_peer_lsn = 0;
g_reply_peer_idx = -1;
xce_vote_reply_hook = mock_vote_reply_hook;
}
/* ── Helper: build a minimal vote endpoint with reply context ────────────── */
static void make_vote_reply_ep(endpoint_t *ep, ha_state_t *ha,
ha_elect_ctx_t *ctx, int peer_idx)
{
memset(ep, 0, sizeof(*ep));
ep->fd = -1;
/* Build and attach a vote_ep_ctx_t via the existing test helper */
vote_ep_ctx_build_for_test(ep, ha, ctx, peer_idx,
atomic_load_explicit(&ha->term, memory_order_acquire),
100ULL, 50ULL, "ctrl-a");
}
/* ── Test: handler decodes correct term, granted, tag, lsn ─────────────── */
static void test_vote_reply_handler_decodes_correct_fields(void)
{
reset_reply_mock();
ha_state_t ha;
ha_init(&ha, &g_vote_cfg, 0);
atomic_store_explicit(&ha.term, 5ULL, memory_order_release);
ha_elect_ctx_t *ctx = alloc_ctx_for_vote(2);
pthread_mutex_init(&ctx->lock, NULL);
endpoint_t ep;
make_vote_reply_ep(&ep, &ha, ctx, 0);
/* Build a real capnp packet with known fields */
io_packet_t *pkt = vote_reply_build_pkt_for_test(
5ULL, /* term */
1, /* vote_granted = true */
200ULL, /* peer_tag */
99ULL /* peer_lsn */);
TEST_ASSERT_NOT_NULL_MESSAGE(pkt, "vote_reply_build_pkt_for_test must succeed");
/* Call handler — it takes ownership of pkt; frees it internally */
int rc = vote_reply_msg_handler_for_test(&ep, pkt);
TEST_ASSERT_EQUAL_INT(0, rc);
/* Verify hook was called exactly once */
TEST_ASSERT_EQUAL_INT_MESSAGE(1, g_reply_hook_calls,
"vote_reply_msg_handler must call hook exactly once");
/* Verify decoded fields */
TEST_ASSERT_EQUAL_INT_MESSAGE(0, g_reply_peer_idx,
"peer_idx must match (0)");
TEST_ASSERT_EQUAL_UINT64_MESSAGE(5ULL, g_reply_term,
"decoded term must be 5");
TEST_ASSERT_EQUAL_INT_MESSAGE(1, g_reply_granted,
"decoded vote_granted must be 1");
TEST_ASSERT_EQUAL_UINT64_MESSAGE(200ULL, g_reply_peer_tag,
"decoded peer_tag must be 200");
TEST_ASSERT_EQUAL_UINT64_MESSAGE(99ULL, g_reply_peer_lsn,
"decoded peer_lsn must be 99");
/* ep->fsm_in.pkt is NOT freed here — handler already freed it */
free(ep.data); /* vote_ep_ctx_t allocated by build helper */
free_ctx_for_vote(ctx);
ha_fini(&ha);
printf("test_vote_reply_handler_decodes_correct_fields: PASS\n");
}
/* ── Test: vote_granted=false correctly decoded ──────────────────────────── */
static void test_vote_reply_handler_denied_vote(void)
{
reset_reply_mock();
ha_state_t ha;
ha_init(&ha, &g_vote_cfg, 0);
atomic_store_explicit(&ha.term, 3ULL, memory_order_release);
ha_elect_ctx_t *ctx = alloc_ctx_for_vote(2);
pthread_mutex_init(&ctx->lock, NULL);
endpoint_t ep;
make_vote_reply_ep(&ep, &ha, ctx, 1);
io_packet_t *pkt = vote_reply_build_pkt_for_test(
3ULL, /* term */
0, /* vote_granted = false */
50ULL, /* peer_tag */
20ULL /* peer_lsn */);
TEST_ASSERT_NOT_NULL_MESSAGE(pkt, "vote_reply_build_pkt_for_test must succeed");
int rc = vote_reply_msg_handler_for_test(&ep, pkt);
TEST_ASSERT_EQUAL_INT(0, rc);
TEST_ASSERT_EQUAL_INT_MESSAGE(1, g_reply_hook_calls,
"hook must be called once even for denied vote");
TEST_ASSERT_EQUAL_INT_MESSAGE(0, g_reply_granted,
"vote_granted must be 0 for denied vote");
TEST_ASSERT_EQUAL_UINT64_MESSAGE(3ULL, g_reply_term,
"term must be decoded correctly even when denied");
TEST_ASSERT_EQUAL_INT_MESSAGE(1, g_reply_peer_idx,
"peer_idx must be 1");
free(ep.data);
free_ctx_for_vote(ctx);
ha_fini(&ha);
printf("test_vote_reply_handler_denied_vote: PASS\n");
}
/* ── Test: term fencing — higher peer term updates ha->term atomically ───── */
static void test_vote_reply_handler_term_fencing(void)
{
reset_reply_mock();
ha_state_t ha;
ha_init(&ha, &g_vote_cfg, 0);
atomic_store_explicit(&ha.term, 3ULL, memory_order_release);
ha_elect_ctx_t *ctx = alloc_ctx_for_vote(2);
pthread_mutex_init(&ctx->lock, NULL);
endpoint_t ep;
make_vote_reply_ep(&ep, &ha, ctx, 0);
/* Peer claims term 10 — higher than our term 3 */
io_packet_t *pkt = vote_reply_build_pkt_for_test(
10ULL, /* peer term > our term */
0, /* denied */
0ULL,
0ULL);
TEST_ASSERT_NOT_NULL_MESSAGE(pkt, "vote_reply_build_pkt_for_test must succeed");
vote_reply_msg_handler_for_test(&ep, pkt);
/* ha->term must now be updated to 10 */
uint64_t updated_term = atomic_load_explicit(&ha.term, memory_order_acquire);
TEST_ASSERT_EQUAL_UINT64_MESSAGE(10ULL, updated_term,
"ha->term must be atomically updated to peer term when peer term > ours");
free(ep.data);
free_ctx_for_vote(ctx);
ha_fini(&ha);
printf("test_vote_reply_handler_term_fencing: PASS\n");
}
/* ── Test: no UAF — pkt is freed before fields are used (ASAN observable) ── */
static void test_vote_reply_handler_no_uaf(void)
{
/* This test proves ASAN-safety: vote_reply_msg_handler reads ALL capnp
* scalars into vr_term/vr_granted/vr_tag/vr_lsn BEFORE calling
* free_packet()+free(pkt). Under ASAN (enabled in xce.build via
* -fsanitize=address), accessing freed memory would fault immediately.
* If this test passes under ASAN, there is no heap-use-after-free. */
reset_reply_mock();
ha_state_t ha;
ha_init(&ha, &g_vote_cfg, 0);
atomic_store_explicit(&ha.term, 2ULL, memory_order_release);
ha_elect_ctx_t *ctx = alloc_ctx_for_vote(2);
pthread_mutex_init(&ctx->lock, NULL);
endpoint_t ep;
make_vote_reply_ep(&ep, &ha, ctx, 0);
io_packet_t *pkt = vote_reply_build_pkt_for_test(
2ULL, 1, 777ULL, 888ULL);
TEST_ASSERT_NOT_NULL_MESSAGE(pkt, "pkt allocation must succeed");
/* Handler frees pkt internally; all field reads happen before the free */
int rc = vote_reply_msg_handler_for_test(&ep, pkt);
TEST_ASSERT_EQUAL_INT(0, rc);
/* Hook captured the fields — verify they are the values we encoded */
TEST_ASSERT_EQUAL_UINT64_MESSAGE(2ULL, g_reply_term, "term matches");
TEST_ASSERT_EQUAL_INT_MESSAGE(1, g_reply_granted, "granted matches");
TEST_ASSERT_EQUAL_UINT64_MESSAGE(777ULL, g_reply_peer_tag, "peer_tag matches");
TEST_ASSERT_EQUAL_UINT64_MESSAGE(888ULL, g_reply_peer_lsn, "peer_lsn matches");
free(ep.data);
free_ctx_for_vote(ctx);
ha_fini(&ha);
printf("test_vote_reply_handler_no_uaf: PASS\n");
}
/* ── Suite entry point ───────────────────────────────────────────────────── */
void test_ha_vote_send(void)
@ -326,4 +565,9 @@ void test_ha_vote_send(void)
RUN_TEST(test_on_vote_connected_encodes_correct_fields);
RUN_TEST(test_on_vote_connected_sets_msg_handler);
RUN_TEST(test_election_timeout_ctx_cancelled_flag);
/* Task 2.3: vote_reply_msg_handler */
RUN_TEST(test_vote_reply_handler_decodes_correct_fields);
RUN_TEST(test_vote_reply_handler_denied_vote);
RUN_TEST(test_vote_reply_handler_term_fencing);
RUN_TEST(test_vote_reply_handler_no_uaf);
}