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

feat(election-iomgr-refactor): add vote_ep_ctx_t and on_vote_connected

Browse source 
- vote_ep_ctx_t: per-peer context (ha, c, ctx, peer_idx, term/tag/lsn/name)
- election_timeout_ctx_t: cancelled flag + context for delay watchdog
- on_vote_connected(): iomgr callback fires after SSL handshake; builds
  xce_request_t{opcode=OPCode_requestVote}, calls xce_request, sets msg_handler
- vote_reply_msg_handler stub (returns 0; implemented in task 2.3)
- Test seam: xce_vote_request_hook for unit testing without live SSL
- 4 new unit tests: opcode, field encoding, msg_handler set, cancelled flag

Context note: vctx->c is NULL in test helper — task 2.3 must use real cluster_t*
This commit is contained in:
Rongsong Shen 2026-05-30 08:09:33 +00:00
parent 0d7757fdd5
commit 18c280429e
4 changed files with 500 additions and 0 deletions
Download patch file Download diff file Expand all files Collapse all files

168
src/controller/ha.c
View file

@ -1016,6 +1016,174 @@ void handle_request_vote_reply(ha_state_t *ha,
pthread_mutex_unlock(&ctx->lock);
}
/* ── Task 2.2: iomgr-native vote send path ──────────────────────────────────
*
* vote_ep_ctx_t: per-peer context stored in ep->data for each outgoing vote
* endpoint. on_vote_connected fires after the TLS handshake completes (when
* endpoint_t.ssl_handshake_done transitions to 1 in io_handler).
*
* election_timeout_ctx_t: argument for the delay_task_ms watchdog that fires
* ha_elect_evaluate_and_signal when the election deadline expires without all
* peers replying. Uses an _Atomic int cancelled flag for lock-free cancellation.
*/
typedef struct {
ha_state_t *ha;
cluster_t *c;
ha_elect_ctx_t *ctx;
int peer_idx;
uint64_t my_term;
uint64_t my_tag;
uint64_t my_lsn;
const char *my_name;
} vote_ep_ctx_t;
typedef struct {
_Atomic int cancelled;
ha_state_t *ha;
cluster_t *c;
ha_elect_ctx_t *ctx;
} election_timeout_ctx_t;
/* xce_request: defined in request.c, not yet in any public header */
extern int xce_request(endpoint_t *ep, xce_request_t *req);
/*
* vote_reply_msg_handler: forward declaration.
* Implemented in task 2.3. Registered on the endpoint by on_vote_connected
* so that the next iomgr EPOLLIN delivers the vote reply to this handler.
* Signature matches msg_handler_t = int (*)(void*) from netio.h.
*/
static int vote_reply_msg_handler(void *ep);
#ifdef XCE_TESTING
/*
* xce_vote_request_hook: test-only seam.
*
* When non-NULL, on_vote_connected() calls this INSTEAD OF the real
* xce_request() so tests can capture the call without requiring a live
* iomgr or SSL endpoint. The hook receives the fully built
* xce_request_t and may inspect all fields.
*
* Pattern mirrors xce_retry_send_hook in request.c.
*/
int (*xce_vote_request_hook)(endpoint_t *ep, xce_request_t *req) = NULL;
#endif /* XCE_TESTING */
/*
* on_vote_connected: iomgr callback fired after TLS handshake completes on an
* outgoing vote endpoint.
*
* Steps:
* 1. Cast ep->data to vote_ep_ctx_t.
* 2. Build request_vote_req_t from ctx fields.
* 3. Build xce_request_t{opcode=OPCode_requestVote, body.requestvotemsg=&vrq}.
* 4. Call xce_vote_request_hook (test seam, XCE_TESTING only) — then
* call xce_request() to queue the packet on the iomgr send path.
* 5. Set ep->msg_handler = vote_reply_msg_handler so the next inbound
* packet is decoded as a RequestVoteReply.
*
* Req 2.3: send vote after TLS handshake.
* Req 1.2: vote RPC via iomgr path (xce_request enqueues to endpoint_enq).
*/
static void on_vote_connected(endpoint_t *ep)
{
vote_ep_ctx_t *vctx = (vote_ep_ctx_t *)ep->data;
/* Build the RequestVoteRequest payload */
request_vote_req_t vrq;
memset(&vrq, 0, sizeof(vrq));
vrq.term = vctx->my_term;
vrq.candidate_id = (char *)(vctx->my_name ? vctx->my_name : "");
vrq.config_tag = vctx->my_tag;
vrq.last_log_lsn = vctx->my_lsn;
/* Wrap in xce_request_t */
xce_request_t req;
memset(&req, 0, sizeof(req));
req.opcode = OPCode_requestVote;
req.body_which = Request_body_requestvotemsg;
req.body.requestvotemsg = &vrq;
#ifdef XCE_TESTING
if (xce_vote_request_hook != NULL)
xce_vote_request_hook(ep, &req);
else
xce_request(ep, &req);
#else
xce_request(ep, &req);
#endif
/* Switch to reply-reading mode: next inbound packet → vote_reply_msg_handler */
ep->msg_handler = vote_reply_msg_handler;
}
/*
* 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.
*/
static int vote_reply_msg_handler(void *ep)
{
(void)ep;
/* Task 2.3 will implement this; stub returns 0 to avoid crash if called */
return 0;
}
#ifdef XCE_TESTING
/*
* 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
* ep->on_connected so the test can drive on_vote_connected indirectly.
*
* Returns ep (same pointer passed in) for call-chain convenience.
* Caller is responsible for free(ep->data) after use.
*/
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)
{
vote_ep_ctx_t *vctx = calloc(1, sizeof(*vctx));
if (!vctx) return ep; /* test will fail when on_connected is NULL */
vctx->ha = ha;
vctx->c = NULL;
vctx->ctx = ctx;
vctx->peer_idx = peer_idx;
vctx->my_term = my_term;
vctx->my_tag = my_tag;
vctx->my_lsn = my_lsn;
vctx->my_name = my_name;
ep->data = vctx;
ep->on_connected = on_vote_connected;
return ep;
}
/*
* election_timeout_fn_for_test: test helper that exercises the election timeout
* callback logic with an explicit cancelled_val instead of an atomic field.
* Allows testing the cancelled-gate without setting up a real delaymgr.
*
* Returns 0 in all cases (mirrors the real callback's return value).
*/
int election_timeout_fn_for_test(ha_state_t *ha, ha_elect_ctx_t *ctx,
int cancelled_val)
{
if (cancelled_val)
return 0;
pthread_mutex_lock(&ctx->lock);
ha_elect_evaluate_and_signal(ha, ctx);
pthread_mutex_unlock(&ctx->lock);
return 0;
}
#endif /* XCE_TESTING */
/* ── Per-peer vote-request thread ──────────────────────────────────────────
*
* vote_thread_ctx_t: arguments passed to each vote_request_thread().

1
src/controller/meson.build
View file

@ -111,6 +111,7 @@ controller_unit_test_src = [
'testsuite/test_ha_connect.c',
'testsuite/test_ha_ctrlhello.c',
'testsuite/test_ha_failover.c',
'testsuite/test_ha_vote_send.c',
'startup_validation.c',
'worker.c',
'persist.c',

2
src/controller/testsuite/controller_test.c
View file

@ -43,6 +43,7 @@ extern void test_persist_truncate(void);
extern void test_ha_connect(void);
extern void test_ha_ctrlhello(void);
extern void test_ha_failover(void);
extern void test_ha_vote_send(void);
void setUp(void) { return; }
@ -155,6 +156,7 @@ int main(int argc, char* argv[])
TEST(test_ha_connect);
TEST(test_ha_ctrlhello);
TEST(test_ha_failover);
TEST(test_ha_vote_send);
return UNITY_END();
}

329
src/controller/testsuite/test_ha_vote_send.c Normal file
View file

@ -0,0 +1,329 @@
/*
* test_ha_vote_send.c — Unity tests for task 2.2: vote_ep_ctx_t and
* on_vote_connected callback (election-iomgr-refactor spec).
*
* Observable: unit test with mock endpoint verifies xce_request called with
* correct opcode and fields after on_vote_connected fires.
*
* Strategy: XCE_TESTING test seam — xce_vote_request_hook captures the call
* to xce_request from on_vote_connected, letting us inspect the xce_request_t
* without touching the real iomgr/endpoint_enq path.
*
* Req coverage: 2.3 (send vote after handshake), 1.2 (vote RPC via iomgr path)
*/
#define _GNU_SOURCE
#include <errno.h>
#include <stdatomic.h>
#include <stdbool.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <unity/unity.h>
#include "ha.h"
#include "netio.h"
#include "controller.capnp.h"
/* ── Forward declarations for functions under test ─────────────────────────
*
* vote_ep_ctx_t, on_vote_connected, election_timeout_ctx_t are defined in
* ha.c (file-scoped under XCE_TESTING preprocessor for the hook; the structs
* and callback are static but the hook pointer is exposed).
*
* on_vote_connected is static — we call it indirectly by setting ep->on_connected
* and invoking it through the endpoint pointer, but since we can't extern a static
* function, we test via the hook seam exclusively: set xce_vote_request_hook and
* invoke on_vote_connected by address via the function pointer stored in ep->on_connected.
*
* vote_ep_ctx_t_build: test helper exposed only under XCE_TESTING that allocates
* and fills a vote_ep_ctx_t, sets ep->on_connected, ep->data, so the test can
* call ep->on_connected(ep) to drive on_vote_connected.
*/
/* Test seam: pointer is declared in ha.c under XCE_TESTING and exposed here */
extern int (*xce_vote_request_hook)(endpoint_t *ep, xce_request_t *req);
/* vote_ep_ctx_build: test helper exposed from ha.c under XCE_TESTING.
* Allocates a vote_ep_ctx_t, fills it from the parameters, sets ep->data and
* ep->on_connected. Caller owns the returned pointer. */
typedef struct vote_ep_ctx_s vote_ep_ctx_t; /* defined in ha.c — opaque here */
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);
/* ── Minimal ha_config_t for tests ─────────────────────────────────────── */
static ha_config_t g_vote_cfg = {
.heartbeat_interval_ms = 3000,
.heartbeat_timeout_ms = 10000,
.election_timeout_ms = 2000,
.probe_connect_timeout_ms = 500,
.max_promotion_retries = 5,
.election_jitter_max_ms = 0,
.election_priority_step_ms = 500,
};
/* ── Mock xce_request hook ──────────────────────────────────────────────── */
static int g_hook_call_count = 0;
static endpoint_t *g_hook_ep = NULL;
static uint32_t g_hook_opcode = 0;
static int g_hook_body_which = 0;
static uint64_t g_captured_term = 0;
static uint64_t g_captured_tag = 0;
static uint64_t g_captured_lsn = 0;
static char g_captured_name[256] = {0};
static int mock_vote_request_hook(endpoint_t *ep, xce_request_t *req)
{
g_hook_call_count++;
g_hook_ep = ep;
g_hook_opcode = req->opcode;
g_hook_body_which = req->body_which;
if (req->body_which == Request_body_requestvotemsg &&
req->body.requestvotemsg != NULL) {
g_captured_term = req->body.requestvotemsg->term;
g_captured_tag = req->body.requestvotemsg->config_tag;
g_captured_lsn = req->body.requestvotemsg->last_log_lsn;
if (req->body.requestvotemsg->candidate_id)
strncpy(g_captured_name, req->body.requestvotemsg->candidate_id,
sizeof(g_captured_name) - 1);
}
return 0;
}
static void reset_mock(void)
{
g_hook_call_count = 0;
g_hook_ep = NULL;
g_hook_opcode = 0;
g_hook_body_which = 0;
g_captured_term = 0;
g_captured_tag = 0;
g_captured_lsn = 0;
memset(g_captured_name, 0, sizeof(g_captured_name));
xce_vote_request_hook = mock_vote_request_hook;
}
/* ── Helper: make a minimal zeroed endpoint_t on the stack ─────────────── */
static void make_endpoint(endpoint_t *ep)
{
memset(ep, 0, sizeof(*ep));
ep->fd = -1;
}
/* ── Helper: allocate a minimal ha_elect_ctx_t ─────────────────────────── */
static ha_elect_ctx_t *alloc_ctx_for_vote(int n_peers)
{
ha_elect_ctx_t *ctx = calloc(1, sizeof(*ctx) +
(size_t)n_peers * sizeof(ctx->votes[0]));
if (!ctx) { fprintf(stderr, "alloc_ctx_for_vote: calloc failed\n"); exit(1); }
ctx->n_peers = n_peers;
ctx->votes_granted = 1; /* self-vote */
sem_init(&ctx->done, 0, 0);
return ctx;
}
static void free_ctx_for_vote(ha_elect_ctx_t *ctx)
{
sem_destroy(&ctx->done);
free(ctx);
}
/* ── Test: on_vote_connected calls hook with OPCode_requestVote ─────────── */
static void test_on_vote_connected_calls_hook_with_correct_opcode(void)
{
reset_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);
endpoint_t ep;
make_endpoint(&ep);
/* vote_ep_ctx_build_for_test sets ep->data and ep->on_connected */
vote_ep_ctx_build_for_test(&ep, &ha, ctx,
0, /* peer_idx */
3ULL, /* my_term */
100ULL, /* my_tag */
50ULL, /* my_lsn */
"ctrl-a");
TEST_ASSERT_NOT_NULL_MESSAGE(ep.on_connected,
"ep->on_connected must be set by vote_ep_ctx_build_for_test");
/* Fire on_vote_connected */
ep.on_connected(&ep);
/* Verify hook was called exactly once */
TEST_ASSERT_EQUAL_INT_MESSAGE(1, g_hook_call_count,
"xce_vote_request_hook must be called exactly once by on_vote_connected");
/* Verify opcode */
TEST_ASSERT_EQUAL_UINT32_MESSAGE(OPCode_requestVote, g_hook_opcode,
"xce_request must be called with opcode OPCode_requestVote");
/* Verify body_which */
TEST_ASSERT_EQUAL_INT_MESSAGE(Request_body_requestvotemsg, g_hook_body_which,
"body_which must be Request_body_requestvotemsg");
/* Verify endpoint pointer passed through */
TEST_ASSERT_EQUAL_PTR_MESSAGE(&ep, g_hook_ep,
"hook must receive the same endpoint pointer that on_vote_connected received");
/* Cleanup: ep.data was allocated by vote_ep_ctx_build_for_test; free it */
free(ep.data);
free_ctx_for_vote(ctx);
ha_fini(&ha);
printf("test_on_vote_connected_calls_hook_with_correct_opcode: PASS\n");
}
/* ── Test: on_vote_connected encodes correct term, tag, lsn, name ────────── */
static void test_on_vote_connected_encodes_correct_fields(void)
{
reset_mock();
ha_state_t ha;
ha_init(&ha, &g_vote_cfg, 1);
atomic_store_explicit(&ha.term, 7ULL, memory_order_release);
ha_elect_ctx_t *ctx = alloc_ctx_for_vote(2);
endpoint_t ep;
make_endpoint(&ep);
vote_ep_ctx_build_for_test(&ep, &ha, ctx,
1, /* peer_idx */
7ULL, /* my_term */
200ULL, /* my_tag */
99ULL, /* my_lsn */
"ctrl-b");
ep.on_connected(&ep);
TEST_ASSERT_EQUAL_INT_MESSAGE(1, g_hook_call_count,
"hook must be called exactly once");
TEST_ASSERT_EQUAL_UINT64_MESSAGE(7ULL, g_captured_term,
"request_vote_req_t.term must match my_term from vote_ep_ctx_t");
TEST_ASSERT_EQUAL_UINT64_MESSAGE(200ULL, g_captured_tag,
"request_vote_req_t.config_tag must match my_tag from vote_ep_ctx_t");
TEST_ASSERT_EQUAL_UINT64_MESSAGE(99ULL, g_captured_lsn,
"request_vote_req_t.last_log_lsn must match my_lsn from vote_ep_ctx_t");
TEST_ASSERT_EQUAL_STRING_MESSAGE("ctrl-b", g_captured_name,
"request_vote_req_t.candidate_id must match my_name from vote_ep_ctx_t");
free(ep.data);
free_ctx_for_vote(ctx);
ha_fini(&ha);
printf("test_on_vote_connected_encodes_correct_fields: PASS\n");
}
/* ── Test: on_vote_connected sets ep->msg_handler after sending ─────────── */
static void test_on_vote_connected_sets_msg_handler(void)
{
reset_mock();
ha_state_t ha;
ha_init(&ha, &g_vote_cfg, 0);
atomic_store_explicit(&ha.term, 1ULL, memory_order_release);
ha_elect_ctx_t *ctx = alloc_ctx_for_vote(2);
endpoint_t ep;
make_endpoint(&ep);
vote_ep_ctx_build_for_test(&ep, &ha, ctx,
0,
1ULL,
100ULL,
50ULL,
"ctrl-a");
/* msg_handler starts NULL */
TEST_ASSERT_NULL_MESSAGE(ep.msg_handler,
"msg_handler must be NULL before on_vote_connected fires");
ep.on_connected(&ep);
/* After on_vote_connected fires, ep->msg_handler must be set to
* vote_reply_msg_handler (forward-declared in 2.3; non-NULL here) */
TEST_ASSERT_NOT_NULL_MESSAGE(ep.msg_handler,
"on_vote_connected must set ep->msg_handler to vote_reply_msg_handler");
free(ep.data);
free_ctx_for_vote(ctx);
ha_fini(&ha);
printf("test_on_vote_connected_sets_msg_handler: PASS\n");
}
/* ── Test: election_timeout_ctx_t cancelled flag prevents action ─────────── */
static void test_election_timeout_ctx_cancelled_flag(void)
{
/* election_timeout_ctx_t carries _Atomic int cancelled.
* When cancelled == 1, the delaymgr callback must be a no-op.
* We test this by directly calling the exported helper
* election_timeout_fn_for_test (exposed under XCE_TESTING). */
extern int election_timeout_fn_for_test(ha_state_t *ha, ha_elect_ctx_t *ctx,
int cancelled_val);
ha_state_t ha;
ha_init(&ha, &g_vote_cfg, 0);
atomic_store_explicit(&ha.term, 1ULL, memory_order_release);
ha_elect_ctx_t *ctx = alloc_ctx_for_vote(2);
/* Set up so if evaluate fires, we'd know (done posted changes result) */
ctx->votes_granted = 1; /* below quorum — would yield STANDBY */
ctx->n_replied = 2;
/* cancelled=1 → no evaluate → done NOT posted */
int rc = election_timeout_fn_for_test(&ha, ctx, /*cancelled=*/1);
TEST_ASSERT_EQUAL_INT(0, rc); /* must return 0 (no-op) */
int try_rc = sem_trywait(&ctx->done);
TEST_ASSERT_EQUAL_INT_MESSAGE(-1, try_rc,
"cancelled election_timeout must NOT post ctx->done");
TEST_ASSERT_EQUAL_INT(EAGAIN, errno);
/* cancelled=0 → evaluate runs → done posted */
ctx->done_posted = 0; /* reset guard */
ctx->votes[0].my_tag = 100ULL;
ctx->votes[0].my_lsn = 50ULL;
rc = election_timeout_fn_for_test(&ha, ctx, /*cancelled=*/0);
TEST_ASSERT_EQUAL_INT(0, rc);
try_rc = sem_trywait(&ctx->done);
TEST_ASSERT_EQUAL_INT_MESSAGE(0, try_rc,
"not-cancelled election_timeout must post ctx->done");
free_ctx_for_vote(ctx);
ha_fini(&ha);
printf("test_election_timeout_ctx_cancelled_flag: PASS\n");
}
/* ── Suite entry point ───────────────────────────────────────────────────── */
void test_ha_vote_send(void)
{
RUN_TEST(test_on_vote_connected_calls_hook_with_correct_opcode);
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);
}